Repository: microsoft/msticpy (language: python)

Issue:
Changing the MDE/M365D request content to json encoding.
Defender API no longer accepts stringified json as a content.
Passing dict of query to json parameter to to httpx.post() (which automatically encodes as json).


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 1c70e11b1..3a347f238 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -53,6 +53,7 @@ jobs:
       - name: Install test dependencies
         run: |
           if [ -f requirements-dev.txt ]; then
+            python -m pip install prospector
             python -m pip install -r requirements-dev.txt
           else
             python -m pip install flake8 black bandit mypy lxml pylint types-attrs
diff --git a/conda/conda-reqs.txt b/conda/conda-reqs.txt
index afcde7669..6290857f1 100644
--- a/conda/conda-reqs.txt
+++ b/conda/conda-reqs.txt
@@ -31,6 +31,7 @@ numpy>=1.15.4
 pandas>=1.4.0
 pygeohash>=1.2.0
 pygments>=2.0.0
+pyjwt>=2.3.0
 python-dateutil>=2.8.1
 pytz>=2019.2
 pyyaml>=3.13
diff --git a/docs/source/api/msticpy.init.azure_synapse_tools.rst b/docs/source/api/msticpy.init.azure_synapse_tools.rst
new file mode 100644
index 000000000..335346cb5
--- /dev/null
+++ b/docs/source/api/msticpy.init.azure_synapse_tools.rst
@@ -0,0 +1,7 @@
+msticpy.init.azure\_synapse\_tools module
+=========================================
+
+.. automodule:: msticpy.init.azure_synapse_tools
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/docs/source/api/msticpy.init.rst b/docs/source/api/msticpy.init.rst
index 5f2f2dccf..d559ac3f5 100644
--- a/docs/source/api/msticpy.init.rst
+++ b/docs/source/api/msticpy.init.rst
@@ -22,6 +22,7 @@ Submodules
    :maxdepth: 4
 
    msticpy.init.azure_ml_tools
+   msticpy.init.azure_synapse_tools
    msticpy.init.mp_pandas_accessors
    msticpy.init.nbinit
    msticpy.init.nbmagics
diff --git a/msticpy/_version.py b/msticpy/_version.py
index 3969427db..db97cd52b 100644
--- a/msticpy/_version.py
+++ b/msticpy/_version.py
@@ -1,2 +1,2 @@
 """Version file."""
-VERSION = "2.0.0"
+VERSION = "2.1.0"
diff --git a/msticpy/auth/azure_auth.py b/msticpy/auth/azure_auth.py
index 32e5189ed..ba944f8d3 100644
--- a/msticpy/auth/azure_auth.py
+++ b/msticpy/auth/azure_auth.py
@@ -21,7 +21,9 @@
 from .azure_auth_core import (  # noqa: F401
     AzCredentials,
     AzureCloudConfig,
+    AzureCredEnvNames,
     az_connect_core,
+    list_auth_methods,
     only_interactive_cred,
 )
 from .cred_wrapper import CredentialWrapper
@@ -34,6 +36,7 @@ def az_connect(
     auth_methods: List[str] = None,
     tenant_id: str = None,
     silent: bool = False,
+    **kwargs,
 ) -> AzCredentials:
     """
     Connect to Azure SDK/API.
@@ -42,21 +45,16 @@ def az_connect(
     ----------
     auth_methods : List[str], optional
         List of authentication methods to try
-        Possible options are:
-        - "env" - to get authentication details from environment variables
-        - "cli" - to use Azure CLI authentication details
-        - "msi" - to user Managed Service Identity details
-        - "interactive" - to prompt for interactive login
-        - "vscode" - to use VSCode credentials
-        - "powershell" - to use PowerShell credentials
-        - "interactive" - to prompt for interactive login
-        - "cache" - to use shared token cache credentials
-        Default is ["env", "msi", "vscode", "cli", "powershell", "interactive"]
+        For a list of possible authentication methods use the `list_auth_methods`
+        function.
+        Default is ["env", "msi", "vscode", "cli", "powershell", "devicecode"]
     tenant_id : str, optional
         The tenant to authenticate against. If not supplied, the default
         tenant for the identity will be used.
     silent : bool, optional
         Set True to hide all output during connection, by default False
+    credential : AzureCredential
+        If an Azure credential is passed, it will be used directly.
 
     Returns
     -------
@@ -71,6 +69,10 @@ def az_connect(
     CloudError
         If chained token credential creation fails.
 
+    See Also
+    --------
+    list_auth_methods
+
     """
     az_cloud_config = AzureCloudConfig()
     # Use auth_methods param or configuration defaults
@@ -85,11 +87,17 @@ def az_connect(
         and isinstance(auth_methods, list)
         and "env" in auth_methods
     ):
-        os.environ["AZURE_CLIENT_ID"] = az_cli_config.args.get("clientId") or ""
-        os.environ["AZURE_TENANT_ID"] = az_cli_config.args.get("tenantId") or ""
-        os.environ["AZURE_CLIENT_SECRET"] = az_cli_config.args.get("clientSecret") or ""
+        os.environ[AzureCredEnvNames.AZURE_CLIENT_ID] = (
+            az_cli_config.args.get("clientId") or ""
+        )
+        os.environ[AzureCredEnvNames.AZURE_TENANT_ID] = (
+            az_cli_config.args.get("tenantId") or ""
+        )
+        os.environ[AzureCredEnvNames.AZURE_CLIENT_SECRET] = (
+            az_cli_config.args.get("clientSecret") or ""
+        )
     credentials = az_connect_core(
-        auth_methods=auth_methods, tenant_id=tenant_id, silent=silent
+        auth_methods=auth_methods, tenant_id=tenant_id, silent=silent, **kwargs
     )
     sub_client = SubscriptionClient(
         credential=credentials.modern,
diff --git a/msticpy/auth/azure_auth_core.py b/msticpy/auth/azure_auth_core.py
index a848858f5..7cb98248e 100644
--- a/msticpy/auth/azure_auth_core.py
+++ b/msticpy/auth/azure_auth_core.py
@@ -5,8 +5,8 @@
 # --------------------------------------------------------------------------
 """Azure KeyVault pre-authentication."""
 
-import contextlib
 import logging
+import os
 import sys
 from collections import namedtuple
 from datetime import datetime
@@ -28,17 +28,12 @@
     VisualStudioCodeCredential,
 )
 from dateutil import parser
-from msrestazure import azure_cloud
 
 from .._version import VERSION
-from ..common import pkg_config as config
 from ..common.exceptions import MsticpyAzureConfigError
-from .cloud_mappings import (
-    CLOUD_ALIASES,
-    CLOUD_MAPPING,
-    get_all_endpoints,
-    get_all_suffixes,
-)
+
+# pylint: disable=unused-import
+from .cloud_mappings import AzureCloudConfig, default_auth_methods  # noqa: F401
 from .cred_wrapper import CredentialWrapper
 
 __version__ = VERSION
@@ -47,118 +42,28 @@
 
 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."""
-    with contextlib.suppress(KeyError):
-        az_settings = config.get_config("Azure")
-        if az_settings and key in az_settings:
-            return az_settings[key]
-    return default
-
-
-def default_auth_methods() -> List[str]:
-    """Get the default (all) authentication options."""
-    return get_azure_config_value("auth_methods", ["cli", "msi", "interactive"])
-
-
-class AzureCloudConfig:
-    """Azure Cloud configuration."""
-
-    def __init__(self, cloud: str = None, tenant_id: Optional[str] = None):
-        """
-        Initialize AzureCloudConfig from `cloud` or configuration.
-
-        Parameters
-        ----------
-        cloud : str, optional
-            The cloud to retrieve configuration for. If not supplied,
-            the cloud ID is read from configuration. If this is not available,
-            it defaults to 'global'.
-        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.
-
-        """
-        self.cloud = cloud or get_azure_config_value("cloud", "global")
-        self.tenant_id = tenant_id or get_azure_config_value("tenant_id", None)
-        self.auth_methods = default_auth_methods()
-
-    @property
-    def cloud_names(self) -> List[str]:
-        """Return a list of current cloud names."""
-        return list(CLOUD_MAPPING.keys())
-
-    @staticmethod
-    def resolve_cloud_alias(alias) -> Optional[str]:
-        """Return match of cloud alias or name."""
-        alias_cf = alias.casefold()
-        aliases = {alias.casefold(): cloud for alias, cloud in CLOUD_ALIASES.items()}
-        if alias_cf in aliases:
-            return aliases[alias_cf]
-        if alias_cf in aliases.values():
-            return alias_cf
-        return None
-
-    @property
-    def endpoints(self) -> azure_cloud.CloudEndpoints:
-        """
-        Get a list of all the endpoints for an Azure cloud.
-
-        Returns
-        -------
-        dict
-            A dictionary of endpoints for the cloud.
-
-        Raises
-        ------
-        MsticpyAzureConfigError
-            If the cloud name is not valid.
-
-        """
-        return get_all_endpoints(self.cloud)
-
-    @property
-    def suffixes(self) -> azure_cloud.CloudSuffixes:
-        """
-        Get a list of all the suffixes for an Azure cloud.
-
-        Returns
-        -------
-        dict
-            A dictionary of suffixes for the cloud.
-
-        Raises
-        ------
-        MsticpyAzureConfigError
-            If the cloud name is not valid.
 
-        """
-        return get_all_suffixes(self.cloud)
+# pylint: disable=too-few-public-methods
+class AzureCredEnvNames:
+    """Enumeration of Azure environment credential names."""
 
-    @property
-    def token_uri(self) -> str:
-        """Return the resource manager token URI."""
-        return f"{self.endpoints.resource_manager}.default"
+    AZURE_CLIENT_ID = "AZURE_CLIENT_ID"  # The app ID for the service principal
+    AZURE_TENANT_ID = "AZURE_TENANT_ID"  # The service principal's Azure AD tenant ID
+    # pylint: disable=line-too-long
+    # [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="This is an enum of env variable names")]
+    AZURE_CLIENT_SECRET = "AZURE_CLIENT_SECRET"  # nosec  # noqa
 
 
-def _build_env_client(
-    tenant_id: str = None, aad_uri: str = None, **kwargs
-) -> EnvironmentCredential:
+def _build_env_client(aad_uri: str = None, **kwargs) -> Optional[EnvironmentCredential]:
     """Build a credential from environment variables."""
     del kwargs
-    return EnvironmentCredential(tenant_id=tenant_id, authority=aad_uri)  # type: ignore
+    if (
+        AzureCredEnvNames.AZURE_CLIENT_ID not in os.environ
+        and AzureCredEnvNames.AZURE_CLIENT_SECRET not in os.environ
+    ):
+        # avoid creating env credential if require envs not set.
+        return None
+    return EnvironmentCredential(authority=aad_uri)  # type: ignore
 
 
 def _build_cli_client(**kwargs) -> AzureCliCredential:
@@ -171,8 +76,13 @@ def _build_msi_client(
     tenant_id: str = None, aad_uri: str = None, **kwargs
 ) -> ManagedIdentityCredential:
     """Build a credential from Managed Identity."""
-    del kwargs
-    return ManagedIdentityCredential(tenant_id=tenant_id, authority=aad_uri)
+    msi_kwargs = kwargs.copy()
+    if AzureCredEnvNames.AZURE_CLIENT_ID in os.environ:
+        msi_kwargs["client_id"] = os.environ[AzureCredEnvNames.AZURE_CLIENT_ID]
+
+    return ManagedIdentityCredential(
+        tenant_id=tenant_id, authority=aad_uri, **msi_kwargs
+    )
 
 
 def _build_vscode_client(
@@ -257,6 +167,11 @@ def _build_powershell_client(**kwargs) -> AzurePowerShellCredential:
 )
 
 
+def list_auth_methods() -> List[str]:
+    """Return list of accepted authentication methods."""
+    return sorted(_CLIENTS.keys())
+
+
 def _az_connect_core(
     auth_methods: List[str] = None,
     cloud: str = None,
@@ -271,14 +186,8 @@ def _az_connect_core(
     ----------
     auth_methods : List[str], optional
         List of authentication methods to try
-        Possible options are:
-        - "env" - to get authentication details from environment variables
-        - "cli" - to use Azure CLI authentication details
-        - "msi" - to user Managed Service Identity details
-        - "vscode" - to use VSCode credentials
-        - "powershell" - to use PowerShell credentials
-        - "interactive" - to prompt for interactive login
-        - "cache" - to use shared token cache credentials
+        For a list of possible authentication methods use the `list_auth_methods`
+        function.
         If not set, it will use the value defined in msticpyconfig.yaml.
         If this is not set, the default is ["env", "cli", "msi", "interactive"]
     cloud : str, optional
@@ -290,6 +199,8 @@ def _az_connect_core(
         the tenant ID is read from configuration, or the default tenant for the identity.
     silent : bool, optional
         Whether to display any output during auth process. Default is False.
+    credential : AzureCredential
+        If an Azure credential is passed, it will be used directly.
 
     Returns
     -------
@@ -303,6 +214,10 @@ def _az_connect_core(
     CloudError
         If chained token credential creation fails.
 
+    See Also
+    --------
+    list_auth_methods
+
     Notes
     -----
     The function tries to obtain credentials from the following
@@ -320,9 +235,14 @@ def _az_connect_core(
     az_config = AzureCloudConfig(cloud)
     aad_uri = az_config.endpoints.active_directory
     tenant_id = tenant_id or AzureCloudConfig().tenant_id
-    creds = _build_chained_creds(
-        aad_uri=aad_uri, requested_clients=auth_methods, tenant_id=tenant_id, **kwargs
-    )
+    creds = kwargs.pop("credential", None)
+    if not creds:
+        creds = _build_chained_creds(
+            aad_uri=aad_uri,
+            requested_clients=auth_methods,
+            tenant_id=tenant_id,
+            **kwargs,
+        )
 
     # Filter and replace error message when credentials not found
     handler = logging.StreamHandler(sys.stdout)
@@ -383,7 +303,7 @@ def _build_chained_creds(
             "At least one valid authentication method required."
         )
 
-    return ChainedTokenCredential(*clients)
+    return ChainedTokenCredential(*[client for client in clients if client])
 
 
 class _AzCachedConnect:
diff --git a/msticpy/auth/cloud_mappings.py b/msticpy/auth/cloud_mappings.py
index 700dc2246..65dcc0a5f 100644
--- a/msticpy/auth/cloud_mappings.py
+++ b/msticpy/auth/cloud_mappings.py
@@ -4,9 +4,13 @@
 # license information.
 # --------------------------------------------------------------------------
 """Azure Cloud Mappings."""
+import contextlib
+from typing import Dict, List, Optional
+
 from msrestazure import azure_cloud
 
 from .._version import VERSION
+from ..common import pkg_config as config
 from ..common.exceptions import MsticpyAzureConfigError
 
 __version__ = VERSION
@@ -50,7 +54,7 @@ def create_cloud_ep_dict(endpoint: str) -> dict:
     Parameters
     ----------
     endpoint : str
-        The name of the endpoint to retreive for each cloud.
+        The name of the endpoint to retrieve for each cloud.
 
     Returns
     -------
@@ -124,3 +128,134 @@ def get_all_suffixes(cloud: str) -> azure_cloud.CloudSuffixes:
         Valid names are 'global', 'usgov', 'de', 'cn'"""
         ) from cloud_err
     return endpoints
+
+
+def get_azure_config_value(key, default):
+    """Get a config value from Azure section."""
+    with contextlib.suppress(KeyError):
+        az_settings = config.get_config("Azure")
+        if az_settings and key in az_settings:
+            return az_settings[key]
+    return default
+
+
+def default_auth_methods() -> List[str]:
+    """Get the default (all) authentication options."""
+    return get_azure_config_value(
+        "auth_methods", ["env", "msi", "vscode", "cli", "powershell", "devicecode"]
+    )
+
+
+class AzureCloudConfig:
+    """Azure Cloud configuration."""
+
+    def __init__(self, cloud: str = None, tenant_id: Optional[str] = None):
+        """
+        Initialize AzureCloudConfig from `cloud` or configuration.
+
+        Parameters
+        ----------
+        cloud : str, optional
+            The cloud to retrieve configuration for. If not supplied,
+            the cloud ID is read from configuration. If this is not available,
+            it defaults to 'global'.
+        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.
+
+        """
+        self.cloud = cloud or get_azure_config_value("cloud", "global")
+        self.tenant_id = tenant_id or get_azure_config_value("tenant_id", None)
+        self.auth_methods = default_auth_methods()
+
+    @property
+    def cloud_names(self) -> List[str]:
+        """Return a list of current cloud names."""
+        return list(CLOUD_MAPPING.keys())
+
+    @staticmethod
+    def resolve_cloud_alias(alias) -> Optional[str]:
+        """Return match of cloud alias or name."""
+        alias_cf = alias.casefold()
+        aliases = {alias.casefold(): cloud for alias, cloud in CLOUD_ALIASES.items()}
+        if alias_cf in aliases:
+            return aliases[alias_cf]
+        return alias_cf if alias_cf in aliases.values() else None
+
+    @property
+    def endpoints(self) -> azure_cloud.CloudEndpoints:
+        """
+        Get the CloudEndpoints class for an Azure cloud.
+
+        Returns
+        -------
+        azure_cloud.CloudEndpoints
+            A CloudEndpoints class for the cloud.
+
+        Raises
+        ------
+        MsticpyAzureConfigError
+            If the cloud name is not valid.
+
+        """
+        return get_all_endpoints(self.cloud)
+
+    @property
+    def suffixes(self) -> azure_cloud.CloudSuffixes:
+        """
+        Get CloudSuffixes class an Azure cloud.
+
+        Returns
+        -------
+        azure_cloud.CloudSuffixes
+            A CloudSuffixes class for the cloud.
+
+        Raises
+        ------
+        MsticpyAzureConfigError
+            If the cloud name is not valid.
+
+        """
+        return get_all_suffixes(self.cloud)
+
+    @property
+    def endpoint(self) -> Dict[str, str]:
+        """
+        Get a dict of all the endpoints for an Azure cloud.
+
+        Returns
+        -------
+        dict
+            A dictionary of endpoints for the cloud.
+
+        Raises
+        ------
+        MsticpyAzureConfigError
+            If the cloud name is not valid.
+
+        """
+        return vars(get_all_endpoints(self.cloud))
+
+    @property
+    def suffix(self) -> azure_cloud.CloudSuffixes:
+        """
+        Get a dict of all the suffixes for an Azure cloud.
+
+        Returns
+        -------
+        dict
+            A dictionary of suffixes for the cloud.
+
+        Raises
+        ------
+        MsticpyAzureConfigError
+            If the cloud name is not valid.
+
+        """
+        return vars(get_all_suffixes(self.cloud))
+
+    @property
+    def token_uri(self) -> str:
+        """Return the resource manager token URI."""
+        return f"{self.endpoints.resource_manager}.default"
diff --git a/msticpy/auth/keyvault_client.py b/msticpy/auth/keyvault_client.py
index 750777175..224f3e784 100644
--- a/msticpy/auth/keyvault_client.py
+++ b/msticpy/auth/keyvault_client.py
@@ -159,14 +159,14 @@ def __init__(
 
     def _try_credential_types(self, **kwargs):
         """Try to access Key Vault to establish usable authentication method."""
-        credentials = kwargs.pop("credential", None)
-        if credentials:
-            kv_client = SecretClient(self.vault_uri, kwargs.pop("credential"))
+        credential = kwargs.pop("credential", None)
+        if credential:
+            kv_client = SecretClient(self.vault_uri, credential=credential)
             try:
                 self._get_working_kv_client(kv_client)
             except ClientAuthenticationError as client_err:
                 _print_status(
-                    f"Could not obtain access token using {credentials.__class__.__name__}."
+                    f"Could not obtain access token using {credential.__class__.__name__}."
                 )
                 self._raise_auth_failed_error(client_err)
 
@@ -175,8 +175,8 @@ def _try_credential_types(self, **kwargs):
                 f"Attempting connection to Key Vault using {auth_method} credentials...",
                 newline=False,
             )
-            credentials = az_connect_core(auth_methods=[auth_method], **kwargs)
-            kv_client = SecretClient(self.vault_uri, credentials.modern)
+            credential = az_connect_core(auth_methods=[auth_method], **kwargs)
+            kv_client = SecretClient(self.vault_uri, credential.modern)
             try:
                 return self._get_working_kv_client(kv_client)
             except ClientAuthenticationError as client_err:
diff --git a/msticpy/context/tilookup.py b/msticpy/context/tilookup.py
index 468115e2f..5fb3122a3 100644
--- a/msticpy/context/tilookup.py
+++ b/msticpy/context/tilookup.py
@@ -16,7 +16,7 @@
 import asyncio
 import warnings
 from collections import ChainMap
-from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Union
+from typing import Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple, Union
 
 import attr
 import nest_asyncio
@@ -384,6 +384,14 @@ def add_provider(
         else:
             self._secondary_providers[name] = provider
 
+    @property
+    def available_query_types(self) -> List[str]:
+        """Return query types available from loaded providers."""
+        query_types: Set[str] = set()
+        for provider in self._providers.values():
+            query_types.update(provider.ioc_query_defs.keys())
+        return sorted(query_types)
+
     # pylint: disable=too-many-arguments
     def lookup_ioc(
         self,
diff --git a/msticpy/init/azure_ml_tools.py b/msticpy/init/azure_ml_tools.py
index 44da3d56c..1932f6c4c 100644
--- a/msticpy/init/azure_ml_tools.py
+++ b/msticpy/init/azure_ml_tools.py
@@ -20,7 +20,7 @@
 )
 
 from .._version import VERSION
-from ..common.pkg_config import refresh_config, _HOME_PATH
+from ..common.pkg_config import _HOME_PATH, refresh_config
 from ..common.utility import search_for_file
 from ..config import MpConfigFile
 
diff --git a/msticpy/init/azure_synapse_tools.py b/msticpy/init/azure_synapse_tools.py
new file mode 100644
index 000000000..8b81207a8
--- /dev/null
+++ b/msticpy/init/azure_synapse_tools.py
@@ -0,0 +1,526 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+"""Help functions for Synapse pipelines notebooks."""
+import os
+import re
+from pathlib import Path
+from typing import Dict, List, Literal, Optional, Union
+
+import httpx
+import jwt
+
+from .._version import VERSION
+from ..auth.azure_auth import AzureCredEnvNames, az_connect
+from ..common import pkg_config as config
+from ..common.pkg_config import get_http_timeout
+from ..common.provider_settings import get_provider_settings
+from ..common.utility import mp_ua_header
+
+
+class SparkUtilsPlaceHolder:
+    """Placeholder for non-spark environments."""
+
+    def __getattr__(self, attrib):
+        """Raise not implemented error."""
+        raise NotImplementedError(
+            "This function can only be used in an Azure Synapse Spark environment."
+        )
+
+
+try:
+    from notebookutils import mssparkutils
+
+    _IN_SPARK = True
+except ImportError:
+    mssparkutils = SparkUtilsPlaceHolder()
+    _IN_SPARK = False
+
+
+__version__ = VERSION
+__author__ = "Ian Hellen"
+
+_LINKED_SERVICES_URL = (
+    "https://{ws_name}.dev.azuresynapse.net/linkedservices?api-version=2020-12-01"
+)
+
+_AZ_NAME_PATTERN = re.compile(r"^https://(?P<name>[^.]+)\.(?P<suffix>.*$)")
+
+
+def is_in_synapse():
+    """Return True if running in Synapse Pipeline."""
+    return os.environ.get("MMLSPARK_PLATFORM_INFO") == "synapse"
+
+
+IdentityType = Literal["managed", "service_principal"]
+
+
+def init_synapse(
+    identity_type: IdentityType = "service_principal",
+    storage_svc_name: Optional[str] = None,
+    tenant_id: Optional[str] = None,
+    cloud: str = "global",
+):
+    """
+    Initialize Synapse Spark notebook pipeline.
+
+    Parameters
+    ----------
+    identity_type : IdentityType, optional
+        The authentication type to be used by MSTICPy, by default "managed".
+        Options are "managed", "service_principal"
+    storage_svc_name : Optional[str], optional
+        Override the default storage linked service name, by default None
+    tenant_id : Optional[str], optional
+        Override the default tenant_id for the Azure authentication, by default None
+    cloud : str
+        Azure cloud - default is "global".
+
+    Raises
+    ------
+    RuntimeError
+        No suitable linked storage service found.
+        Could not mount configuration data container.
+        Could not authenticate to Azure.
+        Could not retrieve secret from Key Vault.
+
+    """
+    if not is_in_synapse():
+        print("Not running in Azure Synapse environment.")
+        return
+    mp_spark = MPSparkUtils()
+    storage_svc = mp_spark.get_storage_service(storage_svc_name)
+    if not storage_svc:
+        raise RuntimeError("No suitable linked storage service found.")
+    print(f"Using linked storage service {storage_svc.name}")
+
+    print("Mounting storage...", end=" ")
+    stor_acct_match = re.match(_AZ_NAME_PATTERN, storage_svc.url)
+    mount_success = mount_container(
+        store_acct_name=stor_acct_match["name"] if stor_acct_match else "",
+        container=mp_spark.container,
+        mount_path=mp_spark.mount_point,
+        linked_service=storage_svc.name,
+    )
+    if not mount_success:
+        raise RuntimeError("Could not mount configuration data container.")
+    print(f"Mounted {storage_svc.url}/{mp_spark.container} at {mp_spark.config_path}")
+
+    # Set msticpy settings
+    print("Configuring msticpy...", end=" ")
+    _configure_mp_settings(mp_spark)
+    print("settings,", end=" ")
+
+    # Set up authentication parameters
+    if identity_type == "managed":
+        _set_mp_azure_settings(auth_method="msi", cloud=cloud)
+        _set_msi_client_id(mp_spark=mp_spark, tenant_id=tenant_id)
+        print("using managed identity,", end=" ")
+    else:
+        # Publish SP creds from KeyVault
+        _set_mp_azure_settings(auth_method="env", cloud=cloud)
+        _set_azure_env_creds(mp_spark=mp_spark, tenant_id=tenant_id)
+        print("using service principal,", end=" ")
+    print()
+
+    # authenticate with MSI or SP to linked KV
+    creds = az_connect()
+    if not creds:
+        raise RuntimeError("Could not authenticate to Azure.")
+    print("Testing authentication...", end=" ")
+
+    # check key retrieval
+    if not _check_kv_key_retrieval():
+        raise RuntimeError("Could not retrieve secret from Key Vault.")
+    print("Key Vault success.")
+    print("Synapse initialization successful")
+
+
+def current_mounts() -> Dict[str, str]:
+    """Return dictionary of current Synapse mount points."""
+    return {mnt.mountPoint: mnt.source for mnt in mssparkutils.fs.mounts()}
+
+
+def mount_container(
+    store_acct_name: str,
+    container: str,
+    mount_path: str,
+    linked_service: str,
+    folder: str = "",
+    cloud: str = "global",
+) -> bool:
+    """
+    Mount Azure file container to Synapse file system.
+
+    Parameters
+    ----------
+    store_acct_name : str
+        Storage account
+    container : str
+        Container name to mount
+    mount_path : str
+        Local path to mount
+    linked_service : str
+        Storage linked service name
+    folder : str, optional
+        Subfolder of container, if any, by default ""
+    cloud : str, optional
+        Azure Cloud, by default "global"
+
+    Returns
+    -------
+    bool
+        True if mounting successful or if mount point already connected.
+        False if mounting failed or mount point connected to a different
+        storage location.
+
+    Raises
+    ------
+    NotImplementedError
+        If using a cloud other than Azure Global (public cloud).
+
+    """
+    if cloud != "global":
+        raise NotImplementedError("Only supports Azure 'global' cloud currently.")
+
+    mount_path = mount_path if mount_path.startswith("/") else f"/{mount_path}"
+    existing_mounts = current_mounts()
+
+    storage_root = "dfs.core.windows.net/"
+    folder = f"{folder}/" if folder else ""
+
+    storage_url = f"abfss://{container}@{store_acct_name}.{storage_root}{folder}"
+    if mount_path in existing_mounts:
+        print(f"path '{mount_path}' already mounted to {existing_mounts[mount_path]}")
+        return existing_mounts[mount_path] == storage_url
+
+    return mssparkutils.fs.mount(
+        storage_url, mount_path, {"linkedService": linked_service}
+    )
+
+
+# pylint: disable=too-few-public-methods
+class LinkedService:
+    """Azure Synapse Linked Service settings."""
+
+    _TYPE_PROPS = "typeProperties"
+
+    class ServiceTypes:
+        """Enumeration of linked service type names."""
+
+        AzureBlobFS = "AzureBlobFS"
+        AzureSqlDW = "AzureSqlDW"
+        AzureFileStorage = "AzureFileStorage"
+        AzureBlobStorage = "AzureBlobStorage"
+        AzureKeyVault = "AzureKeyVault"
+
+    def __init__(self, **kwargs):
+        """Initialize Linked Service instance."""
+        self.res_id: str = kwargs.get("id")
+        self.name: str = kwargs.get("name")
+        self.entry_type: str = kwargs.get("type")
+        self.etag: str = kwargs.get("etag")
+        self.properties: Dict[
+            str, Union[str, Dict[str, Union[Dict, str]]]
+        ] = kwargs.get("properties", {})
+
+    @property
+    def svc_type(self) -> str:
+        """Return type of linked service."""
+        return self.properties.get("type")  # type: ignore
+
+    def __getattr__(self, name: str):
+        """Return sub property `name` from properties or typeProperties."""
+        if name in self.properties:
+            return self.properties[name]
+        if name in self.properties.get(self._TYPE_PROPS, {}):
+            return self.properties.get(self._TYPE_PROPS, {})[name]  # type: ignore
+        raise AttributeError(f"Unknown attribute {name}")
+
+    def __repr__(self):
+        """Return string representation of instance."""
+        return f"LinkedService(name='{self.name}', type='{self.svc_type}')"
+
+    @property
+    def azure_name(self):
+        """Return Azure resource name."""
+        url = self.properties.get(self._TYPE_PROPS, {}).get("url")
+        if not url:
+            url = self.properties.get(self._TYPE_PROPS, {}).get("baseUrl")
+        if not url:
+            return None
+        az_name = re.search(_AZ_NAME_PATTERN, url)
+        return az_name["name"]
+
+
+# Note storage linked service is generated by Synapse workspace creation process
+# pylint: disable=too-few-public-methods
+class SynapseName:
+    """Name mapping to default values."""
+
+    storage_account_prefix = (
+        "adlsforsentinel"  # + last 7 digit of Sentinel workspace id;
+    )
+    key_vault_name_prefix = "kvforsentinel"  # + last 7 digit of ws I’d;
+    kv_linked_service = "Akvlink"
+    sp_client_id_name = "clientid"
+    # pylint: disable=line-too-long
+    # [SuppressMessage("Microsoft.Security", "CS002:SecretInNextLine", Justification="This is an enum of property names")]
+    sp_client_sec_name = "clientsecret"
+    container = "sentinelfiles"
+
+    def __init__(self, workspace_id: str):
+        """Initialize the Synapse Name class."""
+        self._ws_id = workspace_id
+
+    @property
+    def storage_account(self) -> str:
+        """Return default storage account name."""
+        return f"{self.storage_account_prefix}{self._ws_id[-7:]}"
+
+    @property
+    def key_vault(self) -> str:
+        """Return default Key Vault name."""
+        return f"{self.key_vault_name_prefix}{self._ws_id[-7:]}"
+
+
+class MPSparkUtils:
+    """MSTICPy Spark Synapse Utility class."""
+
+    _DEF_MOUNT_POINT = "msticpy"
+
+    def __init__(
+        self, mount_point: Optional[str] = None, container: Optional[str] = None
+    ):
+        """Initialize MPSparkUtils class."""
+        self.linked_services = [
+            LinkedService(**props)
+            for props in _fetch_linked_services(self.workspace_name)
+        ]
+        self._get_workspace_ids()
+        self.mount_point = mount_point or self._DEF_MOUNT_POINT
+        self.container = container or SynapseName.container
+
+    @property
+    def workspace_name(self) -> str:
+        """Return the Synapse workspace name."""
+        return mssparkutils.env.getWorkspaceName()
+
+    @property
+    def fs_mounts(self) -> Dict[str, str]:
+        """Return a dictionary of mount points and targets."""
+        return {mnt.mountPoint: mnt.source for mnt in mssparkutils.fs.mounts()}
+
+    @property
+    def job_id(self):
+        """Return current job ID."""
+        return mssparkutils.env.getJobId()
+
+    @property
+    def config_path(self):
+        """Return mount path for MSTICPy config."""
+        return Path(f"/synfs/{self.job_id}/{self.mount_point}")
+
+    def get_service_of_type(self, svc_type: str) -> Optional[LinkedService]:
+        """
+        Return the first linked service of specific `svc_type`.
+
+        Parameters
+        ----------
+        svc_type : str
+            Service Type (AzureBlobFS, AzureBlobStorage, AzureSqlDW,
+            AzureKeyVault)
+
+        Returns
+        -------
+        Optional[LinkedService]
+            LinkedService instance for the `svc_type`.
+
+        """
+        try:
+            return next(
+                iter(
+                    lnk_svc
+                    for lnk_svc in self.linked_services
+                    if lnk_svc.svc_type == svc_type
+                )
+            )
+        except StopIteration:
+            return None
+
+    def get_all_services_of_type(self, svc_type) -> List[LinkedService]:
+        """
+        Return list of Linked services of `svc_type`.
+
+        Parameters
+        ----------
+        svc_type : str
+            Service Type (AzureBlobFS, AzureBlobStorage, AzureSqlDW,
+            AzureKeyVault)
+
+        Returns
+        -------
+        List[LinkedService]
+            List of LinkedService instances of type `svc_type`.
+
+        """
+        return [
+            lnk_svc for lnk_svc in self.linked_services if lnk_svc.svc_type == svc_type
+        ]
+
+    def get_ws_default_storage(self) -> Optional[LinkedService]:
+        """
+        Return default storage linked service.
+
+        Returns
+        -------
+        Optional[LinkedService]
+            Default storage service.
+
+        """
+        try:
+            return next(
+                iter(
+                    lnk_svc
+                    for lnk_svc in self.linked_services
+                    if lnk_svc.svc_type == LinkedService.ServiceTypes.AzureBlobFS
+                    and "WorkspaceDefaultStorage" in lnk_svc.name
+                )
+            )
+        except StopIteration:
+            return None
+
+    def get_service(self, svc_name: str) -> Optional[LinkedService]:
+        """
+        Return named linked service.
+
+        Returns
+        -------
+        Optional[LinkedService]
+            Named service service.
+
+        """
+        try:
+            return next(
+                iter(
+                    lnk_svc
+                    for lnk_svc in self.linked_services
+                    if lnk_svc.name == svc_name
+                )
+            )
+        except StopIteration:
+            return None
+
+    def get_storage_service(
+        self, linked_svc_name: Optional[str] = None
+    ) -> LinkedService:
+        """Return linked storage service (named) or default storage."""
+        storage_svc: Optional[LinkedService] = None
+        if linked_svc_name:
+            storage_svc = self.get_service(svc_name=linked_svc_name)
+        if not storage_svc:
+            storage_svc = self.get_ws_default_storage()
+            if storage_svc is None:
+                storage_svc = self.get_service_of_type(
+                    svc_type=LinkedService.ServiceTypes.AzureBlobFS
+                )
+        if not storage_svc:
+            raise TypeError("No linked storage service found.")
+        return storage_svc
+
+    def get_kv_secret(self, secret_name: str) -> str:
+        """Return secret from linked service."""
+        kv_service = self.get_service_of_type(LinkedService.ServiceTypes.AzureKeyVault)
+        if not kv_service:
+            raise ValueError("No linked Key Vault service found.")
+        return mssparkutils.credentials.getSecret(
+            kv_service.azure_name, secret_name, kv_service.name
+        )
+
+    def _get_workspace_ids(self):
+        """Retrieve Synapse Managed Identity details from token."""
+        ws_token = mssparkutils.credentials.getToken("Synapse")
+        alg = jwt.get_unverified_header(ws_token)["alg"]
+        decoded_token = jwt.decode(
+            ws_token, algorithms=[alg], options={"verify_signature": False}
+        )
+        self.application_id = decoded_token["appid"]
+        self.tenant_id = decoded_token["tid"]
+        self.object_id = decoded_token["oid"]
+        self.resource_id = decoded_token.get("xms_mirid")
+
+
+def _fetch_linked_services(ws_name: str):
+    """Fetch list of linked services via Azure Synapse API."""
+    token = mssparkutils.credentials.getToken("Synapse")
+    req_headers = {"Authorization": f"Bearer {token}", **mp_ua_header()}
+
+    resp = httpx.get(
+        _LINKED_SERVICES_URL.format(ws_name=ws_name),
+        headers=req_headers,
+        timeout=get_http_timeout(),
+    )
+    return resp.json().get("value")
+
+
+def _set_azure_env_creds(mp_spark: MPSparkUtils, tenant_id: Optional[str] = None):
+    """Publish Service Principal credentials to environment variables."""
+    os.environ[AzureCredEnvNames.AZURE_TENANT_ID] = tenant_id or mp_spark.tenant_id
+    os.environ[AzureCredEnvNames.AZURE_CLIENT_ID] = mp_spark.get_kv_secret(
+        SynapseName.sp_client_id_name
+    )
+    os.environ[AzureCredEnvNames.AZURE_CLIENT_SECRET] = mp_spark.get_kv_secret(
+        SynapseName.sp_client_sec_name
+    )
+
+
+def _set_msi_client_id(mp_spark: MPSparkUtils, tenant_id: Optional[str] = None):
+    """Publish Service Principal credentials to environment variables."""
+    os.environ[AzureCredEnvNames.AZURE_TENANT_ID] = tenant_id or mp_spark.tenant_id
+    os.environ[AzureCredEnvNames.AZURE_CLIENT_ID] = mp_spark.application_id
+
+
+def _set_mp_azure_settings(auth_method: str, cloud: str = "global"):
+    """Configure in-memory settings for MP Azure authn_methods."""
+    az_settings = config.settings.get("Azure")
+    if not az_settings:
+        config.settings["Azure"] = {
+            "auth_methods": [auth_method],
+            "cloud": cloud,
+        }
+    else:
+        curr_methods = az_settings.get("auth_methods", [])
+        if auth_method in curr_methods:
+            curr_methods.remove(auth_method)
+        az_settings["auth_methods"] = [auth_method, *curr_methods]
+
+
+def _check_kv_key_retrieval() -> bool:
+    """Check that we are able to get a key from Key Vault."""
+    ti_settings = get_provider_settings("TIProviders")
+    prov_settings = next(
+        iter(
+            settings
+            for settings in ti_settings.values()
+            if settings.args and "AuthKey" in settings.args
+        )
+    )
+    value = prov_settings.args.get("AuthKey")
+    return value is not None
+
+
+def _configure_mp_settings(mp_spark: MPSparkUtils):
+    """Set msticpyconfig and GeoIPLite dbfolder paths."""
+    os.environ["MSTICPYCONFIG"] = str(
+        mp_spark.config_path.joinpath("msticpyconfig.yaml")
+    )
+    os.environ["MSTICPYHOME"] = str(mp_spark.config_path)
+    config.refresh_config()
+
+    geolite_settings = (
+        config.settings.get("OtherProviders", {}).get("GeoIPLite", {}).get("Args")
+    )
+    if geolite_settings:
+        geolite_settings["DBFolder"] = str(mp_spark.config_path)
diff --git a/msticpy/init/nbinit.py b/msticpy/init/nbinit.py
index 0cbc64572..9843b13ec 100644
--- a/msticpy/init/nbinit.py
+++ b/msticpy/init/nbinit.py
@@ -31,7 +31,7 @@
 from ..auth.azure_auth_core import AzureCliStatus, check_cli_credentials
 from ..common.check_version import check_version
 from ..common.exceptions import MsticpyException, MsticpyUserError
-from ..common.pkg_config import get_config, refresh_config, validate_config, _HOME_PATH
+from ..common.pkg_config import _HOME_PATH, get_config, refresh_config, validate_config
 from ..common.utility import (
     check_and_install_missing_packages,
     check_kwargs,
@@ -42,6 +42,7 @@
 )
 from .azure_ml_tools import check_versions as check_versions_aml
 from .azure_ml_tools import is_in_aml, populate_config_to_mp_config
+from .azure_synapse_tools import init_synapse, is_in_synapse
 from .pivot import Pivot
 from .user_config import load_user_defaults
 
@@ -193,6 +194,8 @@ def _verbose(verbosity: Optional[int] = None) -> int:
 
 current_providers: Dict[str, Any] = {}  # pylint: disable=invalid-name
 
+_SYNAPSE_KWARGS = ["identity_type", "storage_svc_name", "tenant_id", "cloud"]
+
 
 def _pr_output(*args):
     """Output to IPython display or print."""
@@ -299,6 +302,7 @@ def init_notebook(
             "no_config_check",
             "verbosity",
             "verbose",
+            *_SYNAPSE_KWARGS,
         ],
     )
     user_install: bool = kwargs.pop("user_install", False)
@@ -320,6 +324,12 @@ def init_notebook(
             check_version()
             _pr_output(stdout_cap.getvalue())
 
+    if is_in_synapse():
+        synapse_params = {
+            key: val for key, val in kwargs.items() if key in _SYNAPSE_KWARGS
+        }
+        init_synapse(**synapse_params)
+
     # Handle required packages and imports
     _pr_output("Processing imports....")
     imp_ok = _global_imports(
diff --git a/requirements-all.txt b/requirements-all.txt
index 9a9136617..87d127e21 100644
--- a/requirements-all.txt
+++ b/requirements-all.txt
@@ -41,6 +41,7 @@ pandas>=1.4.0
 passivetotal>=2.5.3
 pygeohash>=1.2.0
 pygments>=2.0.0
+pyjwt>=2.3.0
 python-dateutil>=2.8.1  # pandas
 pytz>=2019.2  # pandas
 pyyaml>=3.13
diff --git a/requirements.txt b/requirements.txt
index ce3ec6d9f..bd876145f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -27,6 +27,7 @@ numpy>=1.15.4  # pandas
 pandas>=1.4.0
 pygeohash>=1.2.0
 pygments>=2.0.0
+pyjwt>=2.3.0
 python-dateutil>=2.8.1  # pandas
 pytz>=2019.2  # pandas
 pyyaml>=3.13
diff --git a/tools/toollib/import_analyzer.py b/tools/toollib/import_analyzer.py
index b3b08d99f..7ff2e9e81 100644
--- a/tools/toollib/import_analyzer.py
+++ b/tools/toollib/import_analyzer.py
@@ -53,6 +53,7 @@ def __init__(self):
     "sumologic": "sumologic-sdk",
     "vt": "vt-py",
     "kqlmagic": "KqlmagicCustom",
+    "jwt": "pyjwt",
 }
 
 


Test patch:
diff --git a/tests/init/mssparkutils_fixtures.py b/tests/init/mssparkutils_fixtures.py
new file mode 100644
index 000000000..b44c86937
--- /dev/null
+++ b/tests/init/mssparkutils_fixtures.py
@@ -0,0 +1,241 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+"""Notebook tools MSSparkTools fixure."""
+from collections import namedtuple
+from typing import List
+
+__author__ = "Ian Hellen"
+
+# pylint: disable=redefined-outer-name, invalid-name, too-few-public-methods
+
+MountPointInfo = namedtuple("MountPointInfo", "mountPoint, source, linkedService")
+# MountPointInfo(
+#   mountPoint=/msticpy,
+#   source=abfss://sentinelfiles@mynsynapse.dfs.core.windows.net/,
+#   linkedService=my-synapse1-WorkspaceDefaultStorage
+# )
+
+
+class MSSparkUtils:
+    """Synapse mssparkutils emulator."""
+
+    def __init__(self, data=None):
+        """Initialize sparkutils mock."""
+        self.fs = MsstFileSystem(self)
+        self.env = MsstEnvironment(self)
+        self.credentials = MsstCredentials(self)
+        self.data = data or {}
+
+
+class MsstFileSystem:
+    """Mock for fs sub-module."""
+
+    def __init__(self, container):
+        """Initialize the sub-component."""
+        self.container = container
+
+    def mount(self, source, mountPoint, extraConfigs=None) -> bool:
+        """Mock mounting a storage location."""
+        if self.container.data.get("mount", False):
+            if "mounts" not in self.container.data:
+                self.container.data["mounts"] = []
+            self.container.data["mounts"].append(
+                MountPointInfo(
+                    source=source, mountPoint=mountPoint, linkedService=extraConfigs
+                )
+            )
+        return self.container.data.get("mount", False)
+
+    def mounts(self) -> List[MountPointInfo]:
+        """Return list of mocked mounts."""
+        return self.container.data.get("mounts", [])
+
+    def add_mount(self, mountPoint: str, source: str, linkedService: str):
+        """Add mock mount entry."""
+        curr_mounts = self.container.data.get("mounts")
+        if not curr_mounts:
+            self.container.data["mounts"] = []
+        self.container.data["mounts"].append(
+            MountPointInfo(
+                mountPoint=mountPoint, source=source, linkedService=linkedService
+            )
+        )
+
+
+class MsstEnvironment:
+    """Mock for env submodule."""
+
+    def __init__(self, container):
+        """Initialize the sub-component."""
+        self.container = container
+
+    def getJobId(self) -> str:
+        """Return job ID."""
+        return str(self.container.data.get("jobId", 1))
+
+    def getWorkspaceName(self) -> str:
+        """Return workspace name."""
+        return self.container.data.get("workspace", "default")
+
+
+class MsstCredentials:
+    """Mock for credentials sub-component."""
+
+    def __init__(self, container):
+        """Initialize the sub-component."""
+        self.container = container
+
+    def getSecret(self, akvName, secret, linkedService=""):
+        """Return mocked value from KV."""
+        del linkedService
+        value = self.container.data.get("secrets", {}).get(secret)
+        if not value and akvName:
+            value = self.container.data.get("secrets", {}).get(akvName, {}).get(secret)
+        return value
+
+    def getToken(self, audience, name=""):
+        """Return mocked token."""
+        del audience, name
+        return "TOKEN"
+
+
+class Jwt:
+    """Mock class for PyJWT."""
+
+    def get_unverified_header(self, token):
+        """Return fake token header dict."""
+        del token
+        return {"alg": "RS256"}
+
+    def decode(self, ws_token, algorithms, options=None):
+        """Return decoded token."""
+        del ws_token, algorithms, options
+        return _JWT_TOKEN
+
+
+_JWT_TOKEN = {
+    "aud": "https://vault.azure.net",
+    "iss": "https://sts.windows.net/AAD-GUID/",
+    "iat": 1661881736,
+    "nbf": 1661881736,
+    "exp": 1661968436,
+    "aio": "U29tZSBWYWx1ZQ",
+    "appid": "App-ID",
+    "appidacr": "2",
+    "idp": "https://sts.windows.net/AAD-GUID/",
+    "oid": "App-OID",
+    "rh": "0.U29tZSBWYWx1ZQ-U29tZSBWYWx1ZQ.",
+    "sub": "Sub-GUID",
+    "tid": "AAD-GUID",
+    "uti": "U29tZSBWYWx1ZQ",
+    "ver": "1.0",
+    "xms_az_tm": "azureinfra",
+    "xms_mirid": (
+        "/subscriptions/Sub-GUID/resourcegroups/MyRG/"
+        "providers/Microsoft.Synapse/workspaces/my-synapse1"
+    ),
+}
+
+
+LINKED_SERVICES_RESP = {
+    "value": [
+        {
+            "id": (
+                "/subscriptions/UUID/resourceGroups/MyRG/providers/Microsoft.Synapse"
+                "/workspaces/my-synapse1/linkedservices/my-synapse1-WorkspaceDefaultStorage"
+            ),
+            "name": "my-synapse1-WorkspaceDefaultStorage",
+            "type": "Microsoft.Synapse/workspaces/linkedservices",
+            "etag": "e0029d27-0000-0200-0000-615e09630000",
+            "properties": {
+                "typeProperties": {"url": "https://mynsynapse.dfs.core.windows.net"},
+                "type": "AzureBlobFS",
+                "connectVia": {
+                    "referenceName": "AutoResolveIntegrationRuntime",
+                    "type": "IntegrationRuntimeReference",
+                },
+            },
+        },
+        {
+            "id": (
+                "/subscriptions/UUID/resourceGroups/MyRG/providers/Microsoft.Synapse"
+                "/workspaces/my-synapse1/linkedservices/my-synapse1-WorkspaceDefaultSqlServer"
+            ),
+            "name": "my-synapse1-WorkspaceDefaultSqlServer",
+            "type": "Microsoft.Synapse/workspaces/linkedservices",
+            "etag": "e0029d28-0000-0200-0000-615e09670000",
+            "properties": {
+                "typeProperties": {
+                    "connectionString": (
+                        "Data Source=tcp:my-synapse1.sql.azuresynapse.net,"
+                        "1433;Initial Catalog=@{linkedService().DBName}"
+                    )
+                },
+                "parameters": {"DBName": {"type": "String"}},
+                "type": "AzureSqlDW",
+                "connectVia": {
+                    "referenceName": "AutoResolveIntegrationRuntime",
+                    "type": "IntegrationRuntimeReference",
+                },
+            },
+        },
+        {
+            "id": (
+                "/subscriptions/UUID/resourceGroups/MyRG/providers/Microsoft.Synapse"
+                "/workspaces/my-synapse1/linkedservices/my_synapse_blob"
+            ),
+            "name": "my_synapse_blob",
+            "type": "Microsoft.Synapse/workspaces/linkedservices",
+            "etag": "89028550-0000-0200-0000-62d080970000",
+            "properties": {
+                "annotations": [],
+                "type": "AzureBlobStorage",
+                "typeProperties": {
+                    "connectionString": (
+                        "DefaultEndpointsProtocol=https;"
+                        "AccountName=mysynapsefs;EndpointSuffix=core.windows.net;"
+                    ),
+                },
+                "connectVia": {
+                    "referenceName": "AutoResolveIntegrationRuntime",
+                    "type": "IntegrationRuntimeReference",
+                },
+            },
+        },
+        {
+            "id": (
+                "/subscriptions/UUID/resourceGroups/MyRG/providers/Microsoft.Synapse"
+                "/workspaces/my-synapse1/linkedservices/my_dls_hs"
+            ),
+            "name": "my_dls_hs",
+            "type": "Microsoft.Synapse/workspaces/linkedservices",
+            "etag": "8f024422-0000-0200-0000-62d0a9500000",
+            "properties": {
+                "annotations": [],
+                "type": "AzureBlobFS",
+                "typeProperties": {"url": "https://mysynapsedls.dfs.core.windows.net/"},
+                "connectVia": {
+                    "referenceName": "AutoResolveIntegrationRuntime",
+                    "type": "IntegrationRuntimeReference",
+                },
+            },
+        },
+        {
+            "id": (
+                "/subscriptions/UUID/resourceGroups/MyRG/providers/Microsoft.Synapse"
+                "/workspaces/my-synapse1/linkedservices/KeyVault1"
+            ),
+            "name": "KeyVault1",
+            "type": "Microsoft.Synapse/workspaces/linkedservices",
+            "etag": "2803ab6a-0000-0200-0000-62d5f6480000",
+            "properties": {
+                "annotations": [],
+                "type": "AzureKeyVault",
+                "typeProperties": {"baseUrl": "https://my-synapse.vault.azure.net/"},
+            },
+        },
+    ]
+}
diff --git a/tests/init/test_azure_synapse_tools.py b/tests/init/test_azure_synapse_tools.py
new file mode 100644
index 000000000..0aa60270e
--- /dev/null
+++ b/tests/init/test_azure_synapse_tools.py
@@ -0,0 +1,350 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+"""Azure Synapse tools unit test."""
+import os
+import re
+from collections import namedtuple
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+import pytest_check as check
+import respx
+
+from msticpy.common import pkg_config
+from msticpy.init import azure_synapse_tools as synapse
+
+from ..unit_test_lib import custom_mp_config, get_test_data_path
+from .mssparkutils_fixtures import LINKED_SERVICES_RESP, Jwt, MSSparkUtils
+
+__author__ = "Ian Hellen"
+
+# pylint: disable=redefined-outer-name, protected-access
+
+
+SparkUtilsTest = namedtuple("SparkUtilsTest", "mp_spark, msspark_mock")
+
+
+@pytest.fixture
+@respx.mock
+def mpsparkutils(monkeypatch):
+    url_pattern = re.compile(
+        r"https://.*dev\.azuresynapse\.net/linkedservices\?api-version=2020-12-01"
+    )
+    respx.get(url_pattern).respond(status_code=200, json=LINKED_SERVICES_RESP)
+
+    spark_data = {
+        "mount": True,
+        "jobId": "2",
+        "workspace": "test-workspace",
+        "secrets": {
+            "special-sec": "the-value",
+            "vault1": {"special-sec": "the-value2"},
+            "clientid": "[PLACEHOLDER]",
+            "clientsecret": "[PLACEHOLDER]",
+        },
+    }
+    spark_mock = MSSparkUtils(data=spark_data)
+    monkeypatch.setattr(synapse, "mssparkutils", spark_mock)
+    monkeypatch.setattr(synapse, "jwt", Jwt())
+
+    return SparkUtilsTest(synapse.MPSparkUtils(), spark_mock)
+
+
+def test_create_mp_spark_utils_instance(mpsparkutils):
+    """Function_docstring."""
+    mp_spark = mpsparkutils.mp_spark
+    check.equal(len(mp_spark.linked_services), 5)
+    svc_types = [ls.svc_type for ls in mp_spark.linked_services]
+    check.is_in("AzureBlobFS", svc_types)
+    check.is_in("AzureSqlDW", svc_types)
+    check.is_in("AzureKeyVault", svc_types)
+
+    check.is_false(mp_spark.fs_mounts)
+    check.equal(mp_spark.job_id, "2")
+    check.equal(mp_spark.config_path, Path(f"/synfs/{mp_spark.job_id}/msticpy"))
+    check.equal(mp_spark.application_id, "App-ID")
+    check.equal(mp_spark.object_id, "App-OID")
+    check.equal(mp_spark.tenant_id, "AAD-GUID")
+    check.equal(mp_spark.mount_point, "msticpy")
+    check.equal(mp_spark.container, "sentinelfiles")
+
+    storage_svc = mp_spark.get_storage_service()
+    check.equal(storage_svc.name, "my-synapse1-WorkspaceDefaultStorage")
+    check.equal(
+        mp_spark.get_storage_service("my-synapse1-WorkspaceDefaultStorage").name,
+        "my-synapse1-WorkspaceDefaultStorage",
+    )
+    check.equal(mp_spark.get_service_of_type("AzureKeyVault").name, "KeyVault1")
+    check.equal(
+        mp_spark.get_service_of_type("AzureSqlDW").name,
+        "my-synapse1-WorkspaceDefaultSqlServer",
+    )
+    check.is_none(mp_spark.get_service_of_type("InvalidType"))
+
+    check.equal(len(mp_spark.get_all_services_of_type("AzureBlobFS")), 2)
+
+    check.equal(
+        mp_spark.get_ws_default_storage().name, "my-synapse1-WorkspaceDefaultStorage"
+    )
+    check.equal(mp_spark.get_service("KeyVault1").svc_type, "AzureKeyVault")
+    check.is_none(mp_spark.get_service("InvalidName"))
+
+
+def test_linked_service_attribs(mpsparkutils):
+    """Test getattr functionality of LinkedService class."""
+    mp_spark = mpsparkutils.mp_spark
+    storage_svc = mp_spark.get_storage_service()
+    check.equal(storage_svc.name, "my-synapse1-WorkspaceDefaultStorage")
+
+    check.equal(storage_svc.azure_name, "mynsynapse")
+    check.is_in(
+        "linkedservices/my-synapse1-WorkspaceDefaultStorage", storage_svc.res_id
+    )
+    check.equal(storage_svc.svc_type, "AzureBlobFS")
+    # check can retrieve subkeys of properties via direct attrib name
+    check.is_instance(storage_svc.typeProperties, dict)
+    # check can retrieve subkeys of typeProperties via direct attrib name
+    check.equal(storage_svc.url, "https://mynsynapse.dfs.core.windows.net")
+
+    with pytest.raises(AttributeError):
+        test = storage_svc.unknown_property
+
+    check.is_in(
+        "LinkedService(name='my-synapse1-WorkspaceDefaultStorage'", repr(storage_svc)
+    )
+
+
+def test_mp_spark_utils_get_secret(mpsparkutils):
+    """Test get_kv_secret function."""
+    mp_spark = mpsparkutils.mp_spark
+
+    check.equal(mp_spark.job_id, "2")
+    check.equal(mp_spark.get_kv_secret("special-sec"), "the-value")
+
+
+def test_mp_spark_utils_mount(mpsparkutils):
+    """Test get_kv_secret function."""
+    mp_spark = mpsparkutils.mp_spark
+    mpsparkutils.msspark_mock.fs.add_mount(
+        "testf", "storage_url", linkedService="my_dls_hs"
+    )
+    check.equal(len(mp_spark.fs_mounts), 1)
+    storage_svc = mp_spark.get_storage_service()
+
+    print("Mounting storage...", end=" ")
+    stor_acct_match = re.match(synapse._AZ_NAME_PATTERN, storage_svc.url)
+    mount_success = synapse.mount_container(
+        store_acct_name=stor_acct_match["name"] if stor_acct_match else "",
+        container=mp_spark.container,
+        mount_path=mp_spark.mount_point,
+        linked_service=storage_svc.name,
+    )
+    check.equal(len(mp_spark.fs_mounts), 2)
+    check.is_true(mount_success)
+
+    # try to mount again - should not add new mount point
+    mount_success = synapse.mount_container(
+        store_acct_name=stor_acct_match["name"] if stor_acct_match else "",
+        container=mp_spark.container,
+        mount_path=mp_spark.mount_point,
+        linked_service=storage_svc.name,
+    )
+    check.equal(len(mp_spark.fs_mounts), 2)
+    check.is_true(mount_success)
+
+    # try to mount same point with different storage - should not add new mount point
+    mount_success = synapse.mount_container(
+        store_acct_name=stor_acct_match["name"] if stor_acct_match else "",
+        container="newcontainer",
+        mount_path=mp_spark.mount_point,
+        linked_service=storage_svc.name,
+    )
+    check.equal(len(mp_spark.fs_mounts), 2)
+    check.is_false(mount_success)
+
+    # try to mount same point with different storage - should not add new mount point
+    mount_success = synapse.mount_container(
+        store_acct_name=stor_acct_match["name"] if stor_acct_match else "",
+        container="newcontainer",
+        mount_path="newmount",
+        linked_service=storage_svc.name,
+    )
+    check.equal(len(mp_spark.fs_mounts), 3)
+    check.is_true(mount_success)
+
+
+def test_synapse_name():
+    """Test the Synapse Name class."""
+    syn_name = synapse.SynapseName(workspace_id="9999-99999-1234567")
+    check.equal(syn_name.storage_account_prefix, "adlsforsentinel")
+    check.equal(syn_name.key_vault_name_prefix, "kvforsentinel")
+    check.equal(syn_name.kv_linked_service, "Akvlink")
+    check.equal(syn_name.sp_client_id_name, "clientid")
+    check.equal(syn_name.sp_client_sec_name, "clientsecret")
+    check.equal(syn_name.container, "sentinelfiles")
+
+    check.equal(syn_name.storage_account, "adlsforsentinel1234567")
+    check.equal(syn_name.key_vault, "kvforsentinel1234567")
+
+    syn_name = synapse.SynapseName(workspace_id="4567")
+    check.equal(syn_name.storage_account, "adlsforsentinel4567")
+    check.equal(syn_name.key_vault, "kvforsentinel4567")
+
+
+def test_set_azure_env_creds(mpsparkutils):
+    """Test setting environment variables for env credential."""
+    mp_spark = mpsparkutils.mp_spark
+    az_cred = synapse.AzureCredEnvNames
+    synapse._set_azure_env_creds(mp_spark)
+    # save any existing variables
+    saved_env = {
+        az_cred.AZURE_TENANT_ID: os.environ.get(az_cred.AZURE_TENANT_ID),
+        az_cred.AZURE_CLIENT_ID: os.environ.get(az_cred.AZURE_CLIENT_ID),
+        az_cred.AZURE_CLIENT_SECRET: os.environ.get(az_cred.AZURE_CLIENT_SECRET),
+    }
+    try:
+        check.equal(os.environ[az_cred.AZURE_TENANT_ID], mp_spark.tenant_id)
+        check.equal(os.environ[az_cred.AZURE_CLIENT_ID], "[PLACEHOLDER]")
+        check.equal(
+            os.environ[az_cred.AZURE_CLIENT_SECRET],
+            "[PLACEHOLDER]",
+        )
+    finally:
+        # restore any changed variables
+        for env, val in saved_env.items():
+            if val:
+                os.environ[env] = val
+
+
+def test_set_azure_msi_creds(mpsparkutils):
+    """Test setting environment variables for env credential."""
+    mp_spark = mpsparkutils.mp_spark
+    az_cred = synapse.AzureCredEnvNames
+    synapse._set_msi_client_id(mp_spark)
+    # save any existing variables
+    saved_env = {
+        az_cred.AZURE_TENANT_ID: os.environ.get(az_cred.AZURE_TENANT_ID),
+        az_cred.AZURE_CLIENT_ID: os.environ.get(az_cred.AZURE_CLIENT_ID),
+    }
+    try:
+        check.equal(os.environ[az_cred.AZURE_TENANT_ID], mp_spark.tenant_id)
+        check.equal(os.environ[az_cred.AZURE_CLIENT_ID], "App-ID")
+    finally:
+        # restore any changed variables
+        for env, val in saved_env.items():
+            if val:
+                os.environ[env] = val
+
+
+def test_set_mp_azure_settings():
+    """Test setting msticpy settings."""
+    mp_config = get_test_data_path().parent.joinpath("msticpyconfig-test.yaml")
+    with custom_mp_config(mp_config):
+        prev_auth_methods = pkg_config.get_config("Azure").get("auth_methods").copy()
+        synapse._set_mp_azure_settings("env")
+        auth_methods = pkg_config.get_config("Azure").get("auth_methods")
+        check.equal(auth_methods[0], "env")
+        check.not_equal(auth_methods, prev_auth_methods)
+
+    with custom_mp_config(mp_config):
+        prev_auth_methods = pkg_config.get_config("Azure").get("auth_methods").copy()
+        del pkg_config.settings["Azure"]
+        synapse._set_mp_azure_settings("env")
+        auth_methods = pkg_config.get_config("Azure").get("auth_methods")
+        check.equal(auth_methods[0], "env")
+        check.not_equal(auth_methods, prev_auth_methods)
+
+    with custom_mp_config(mp_config):
+        prev_auth_methods = pkg_config.get_config("Azure").get("auth_methods").copy()
+        synapse._set_mp_azure_settings(prev_auth_methods[-1])
+        auth_methods = pkg_config.get_config("Azure").get("auth_methods")
+        check.equal(auth_methods[0], prev_auth_methods[-1])
+        check.not_equal(auth_methods, prev_auth_methods)
+
+
+@respx.mock
+def test_init_synapse(mpsparkutils, monkeypatch):
+    """Test the init_synapse function."""
+    url_pattern = re.compile(
+        r"https://.*dev\.azuresynapse\.net/linkedservices\?api-version=2020-12-01"
+    )
+    respx.get(url_pattern).respond(status_code=200, json=LINKED_SERVICES_RESP)
+
+    # We need to patch out some of the functions to
+    # avoid trying to authenticate and connect to Key Vault.
+    config_mp = Mock(return_value=None)
+    monkeypatch.setattr(synapse, "_configure_mp_settings", config_mp)
+    az_connect = Mock(return_value="creds")
+    monkeypatch.setattr(synapse, "az_connect", az_connect)
+    check_kv = Mock(return_value=True)
+    monkeypatch.setattr(synapse, "_check_kv_key_retrieval", check_kv)
+    set_az_settings = Mock(return_value=None)
+    monkeypatch.setattr(synapse, "_set_mp_azure_settings", set_az_settings)
+    set_msi_client = Mock(return_value=None)
+    monkeypatch.setattr(synapse, "_set_msi_client_id", set_msi_client)
+    set_azure_env_creds = Mock(return_value=None)
+    monkeypatch.setattr(synapse, "_set_azure_env_creds", set_azure_env_creds)
+    is_in_synapse = Mock(return_value=True)
+    monkeypatch.setattr(synapse, "is_in_synapse", is_in_synapse)
+
+    synapse.init_synapse(identity_type="managed")
+
+    check.is_instance(config_mp.mock_calls[0].args[0], synapse.MPSparkUtils)
+    check.equal(len(az_connect.mock_calls), 1)
+    check.equal(len(az_connect.mock_calls[0].args), 0)
+    check.equal(len(check_kv.mock_calls), 1)
+    check.equal(len(check_kv.mock_calls[0].args), 0)
+    check.equal(len(set_az_settings.mock_calls), 1)
+    check.equal(len(set_msi_client.mock_calls), 1)
+    check.is_instance(
+        set_msi_client.mock_calls[0].kwargs["mp_spark"], synapse.MPSparkUtils
+    )
+    check.is_instance(
+        set_msi_client.mock_calls[0].kwargs["tenant_id"], (str, type(None))
+    )
+    check.equal(len(set_azure_env_creds.mock_calls), 0)
+    check.equal(set_az_settings.mock_calls[0].kwargs["auth_method"], "msi")
+    check.equal(set_az_settings.mock_calls[0].kwargs["cloud"], "global")
+
+    synapse.init_synapse(identity_type="service_principal")
+    check.is_instance(config_mp.mock_calls[1].args[0], synapse.MPSparkUtils)
+    check.equal(len(az_connect.mock_calls), 2)
+    check.equal(len(set_msi_client.mock_calls), 1)
+    check.equal(len(set_azure_env_creds.mock_calls), 1)
+    check.is_instance(
+        set_azure_env_creds.mock_calls[0].kwargs["mp_spark"], synapse.MPSparkUtils
+    )
+    check.is_instance(
+        set_azure_env_creds.mock_calls[0].kwargs["tenant_id"], (str, type(None))
+    )
+    check.equal(set_az_settings.mock_calls[1].kwargs["auth_method"], "env")
+    check.equal(set_az_settings.mock_calls[0].kwargs["cloud"], "global")
+
+    az_connect.return_value = None
+    with pytest.raises(RuntimeError) as no_connect_err:
+        synapse.init_synapse(identity_type="service_principal")
+        check.equal(no_connect_err.value.args[0], "Could not authenticate to Azure.")
+
+    az_connect.return_value = "creds"
+    check_kv.return_value = False
+    with pytest.raises(RuntimeError) as kv_err:
+        synapse.init_synapse(identity_type="service_principal")
+        check.equal(kv_err.value.args[0], "Could not retrieve secret from Key Vault.")
+
+    prev_calls = len(config_mp.mock_calls)
+    # This should cause early exit from function
+    is_in_synapse.return_value = False
+    synapse.init_synapse(identity_type="service_principal")
+    check.equal(len(config_mp.mock_calls), prev_calls)
+
+
+def test_is_in_synapse(monkeypatch):
+    """Test check for environment variable."""
+    check.is_false(synapse.is_in_synapse())
+
+    monkeypatch.setenv("MMLSPARK_PLATFORM_INFO", "synapse")
+    check.is_true(synapse.is_in_synapse())