Add Sentinel certificate auth support and docs

docs/source/data_acquisition/DataProv-MSSentinel.rst

@@ -215,6 +215,73 @@ You can override several authentication parameters including:
 * auth_types - a list of authentication types to try in order
 * tenant_id - the Azure tenant ID to use for authentication
 
+Using Certificate-based Service Principal Authentication
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Microsoft Sentinel QueryProvider connections support certificate-based
+service principal authentication.
+
+There are two supported ways to use this:
+
+1. Use ``auth_types=["env"]`` and set the Azure SDK certificate
+   environment variables described in
+   :doc:`Azure Authentication <../getting_started/AzureAuthentication>`.
+2. Use ``auth_types=["certificate"]`` and pass the certificate
+   parameters directly to ``connect``.
+
+The environment-variable route is the simplest if you already have a
+working :py:func:`msticpy.auth.azure_auth.az_connect` certificate flow.
+
+.. code:: python
+
+    qry_prov = QueryProvider("MSSentinel")
+    qry_prov.connect(workspace="Default", auth_types=["env"])
+
+If you want to pass the certificate details directly, use Azure SDK
+parameter names:
+
+.. code:: python
+
+    qry_prov = QueryProvider("MSSentinel")
+    qry_prov.connect(
+        workspace="Default",
+        auth_types=["certificate"],
+        tenant_id="<tenant-id>",
+        client_id="<app-id>",
+        certificate_path="/path/to/sp-cert.pem",
+    )
+
+For password-protected PFX certificates, add ``password="<pfx-password>"``.
+If your tenant requires subject name/issuer authentication, add
+``send_certificate_chain=True``.
+
+If you want to keep these settings in ``msticpyconfig.yaml`` for a
+workspace, put them under the workspace ``Args`` section. These values
+are passed through to ``az_connect`` when the provider connects.
+
+.. code:: yaml
+
+    AzureSentinel:
+      Workspaces:
+        Default:
+          WorkspaceId: 271f17d3-5457-4237-9131-ae98a6f55c37
+          TenantId: 335b56ab-67a2-4118-ac14-6eb454f350af
+          Args:
+            client_id: 00000000-0000-0000-0000-000000000000
+            certificate_path: /path/to/sp-cert.pem
+            password:
+              EnvironmentVar: AZURE_CLIENT_CERTIFICATE_PASSWORD
+
+For compatibility, MSTICPy also normalizes common config names such as
+``ClientId``, ``Certificate`` and ``CertificatePassword`` when reading
+workspace ``Args``.
+
+If you see a log message like
+``'certificate' credential requested but client_id param not supplied``
+it usually means that the provider received the certificate request but
+did not receive a ``client_id`` value in the format expected by the
+Azure identity credential builder.
+
 If you are using a Sovereign cloud rather than the Azure global cloud,
 you should follow the guidance in
 :doc:`Azure Authentication <../getting_started/AzureAuthentication>`

docs/source/getting_started/AzureAuthentication.rst

@@ -164,6 +164,41 @@ AZURE_TENANT_ID	      id of the application's Azure Active Directory tenant
 AZURE_CLIENT_SECRET	  one of the application's client secrets
 ==================== ========================================================
 
+To use AppID and certificate authentication, set these environment variables
+
+======================================= ========================================================
+variable name	                         value
+======================================= ========================================================
+AZURE_CLIENT_ID	                         id of an Azure Active Directory application
+AZURE_TENANT_ID	                         id of the application's Azure Active Directory tenant
+AZURE_CLIENT_CERTIFICATE_PATH	         path to a PEM or PFX certificate containing the private key
+AZURE_CLIENT_CERTIFICATE_PASSWORD	     optional password for a PFX certificate
+AZURE_CLIENT_SEND_CERTIFICATE_CHAIN	   optional boolean (`true`/`1`) to send the x5c certificate chain
+======================================= ========================================================
+
+Using Certificate Credentials Directly
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+**auth_method = "certificate"**
+
+If you don't want to rely on environment variables, you can pass the
+certificate settings directly to :py:func:`msticpy.auth.azure_auth.az_connect`.
+
+.. code:: python
+
+  from msticpy.auth.azure_auth import az_connect
+
+  creds = az_connect(
+    auth_methods=["certificate"],
+    tenant_id="<tenant-id>",
+    client_id="<app-id>",
+    certificate_path="/path/to/sp-cert.pem",
+  )
+
+For password-protected PFX certificates, add the ``password`` keyword
+argument. If your tenant requires subject name/issuer authentication,
+you can also pass ``send_certificate_chain=True``.
+
 Using Device-code Credentials
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -221,6 +256,9 @@ include:
 - The :py:meth:`QueryProvider.connect <msticpy.data.core.data_providers.QueryProvider.connect>`
   for Azure data services (such as Microsoft Sentinel)
 
+For :py:meth:`QueryProvider.connect <msticpy.data.core.data_providers.QueryProvider.connect>`
+the equivalent parameter name is ``auth_types``.
+
 Specify the list of one or more ``auth_methods`` that you want to use
 as a list of strings. The authentication methods will be tried in
 the order specified in the list.

msticpy/data/drivers/azure_monitor_driver.py

@@ -28,7 +28,9 @@
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline.policies import UserAgentPolicy
 from packaging.version import Version  # pylint: disable=no-name-in-module
-from packaging.version import parse as parse_version  # pylint: disable=no-name-in-module
+from packaging.version import (
+    parse as parse_version,
+)  # pylint: disable=no-name-in-module
 
 from ..._version import VERSION
 from ...auth.azure_auth import AzureCloudConfig, az_connect
@@ -134,9 +136,13 @@ def __init__(self, connection_str: str | None = None, **kwargs):
             "data_environments",
             ("MSSentinel", "LogAnalytics", "AzureSentinel"),
         )
-        self.set_driver_property(DriverProps.EFFECTIVE_ENV, DataEnvironment.MSSentinel.name)
+        self.set_driver_property(
+            DriverProps.EFFECTIVE_ENV, DataEnvironment.MSSentinel.name
+        )
         self.set_driver_property(DriverProps.SUPPORTS_THREADING, value=True)
-        self.set_driver_property(DriverProps.MAX_PARALLEL, value=kwargs.get("max_threads", 4))
+        self.set_driver_property(
+            DriverProps.MAX_PARALLEL, value=kwargs.get("max_threads", 4)
+        )
         self.az_cloud_config = AzureCloudConfig()
         logger.info(
             "AzureMonitorDriver loaded. connect_str  %s, kwargs: %s",
@@ -312,7 +318,9 @@ def query(
         return data if data is not None else result
 
     # pylint: disable=too-many-branches
-    def query_with_results(self, query: str, **kwargs) -> tuple[pd.DataFrame, dict[str, Any]]:
+    def query_with_results(
+        self, query: str, **kwargs
+    ) -> tuple[pd.DataFrame, dict[str, Any]]:
         """
         Execute query string and return DataFrame of results.
 
@@ -338,7 +346,9 @@ def query_with_results(self, query: str, **kwargs) -> tuple[pd.DataFrame, dict[s
         workspace_id = next(iter(self._workspace_ids), None) or self._workspace_id
         additional_workspaces = self._workspace_ids[1:] if self._workspace_ids else None
         logger.info("Query to run %s", query)
-        logger.info("Workspaces %s", ",".join(self._workspace_ids) or self._workspace_id)
+        logger.info(
+            "Workspaces %s", ",".join(self._workspace_ids) or self._workspace_id
+        )
         logger.info(
             "Time span %s - %s",
             str(time_span_value[0]) if time_span_value else "none",
@@ -399,7 +409,9 @@ def _create_query_client(self, connection_str, **kwargs):
         # check for additional Args in settings but allow kwargs to override
         connect_args = self._get_workspace_settings_args()
         connect_args.update(kwargs)
-        connect_args.update({"auth_methods": az_auth_types, "tenant_id": self._az_tenant_id})
+        connect_args.update(
+            {"auth_methods": az_auth_types, "tenant_id": self._az_tenant_id}
+        )
         credentials = az_connect(**connect_args)
         logger.info(
             "Created query client. Auth type: %s, Url: %s, Proxies: %s",
@@ -419,7 +431,20 @@ def _get_workspace_settings_args(self) -> dict[str, Any]:
             return {}
         args_path = f"{self._ws_config.settings_path}.Args"
         args_settings = self._ws_config.settings.get("Args", {})
-        return {name: get_protected_setting(args_path, name) for name in args_settings.keys()}
+        arg_name_map = {
+            "clientid": "client_id",
+            "clientsecret": "client_secret",
+            "certificate": "certificate_path",
+            "certificatepath": "certificate_path",
+            "certificatepassword": "password",
+            "sendcertificatechain": "send_certificate_chain",
+        }
+        return {
+            arg_name_map.get(name.casefold(), name): get_protected_setting(
+                args_path, name
+            )
+            for name in args_settings.keys()
+        }
 
     def _get_workspaces(self, connection_str: str | None = None, **kwargs):
         """Get workspace or workspaces to connect to."""
@@ -438,12 +463,16 @@ def _get_workspaces(self, connection_str: str | None = None, **kwargs):
         connection_str = connection_str or self._def_connection_str
         if workspace_name or connection_str is None:
             ws_config = WorkspaceConfig(workspace=workspace_name)
-            logger.info("WorkspaceConfig created from workspace name %s", workspace_name)
+            logger.info(
+                "WorkspaceConfig created from workspace name %s", workspace_name
+            )
         elif isinstance(connection_str, str):
             self._def_connection_str = connection_str
             with contextlib.suppress(ValueError):
                 ws_config = WorkspaceConfig.from_connection_string(connection_str)
-                logger.info("WorkspaceConfig created from connection_str %s", connection_str)
+                logger.info(
+                    "WorkspaceConfig created from connection_str %s", connection_str
+                )
         elif isinstance(connection_str, WorkspaceConfig):
             logger.info("WorkspaceConfig as parameter %s", connection_str.workspace_id)
             ws_config = connection_str
@@ -487,9 +516,9 @@ def _get_workspaces_by_id(self, workspace_ids):
 
     def _get_workspaces_by_name(self, workspaces):
         workspace_configs = {
-            WorkspaceConfig(workspace)[WorkspaceConfig.CONF_WS_ID]: WorkspaceConfig(workspace)[
-                WorkspaceConfig.CONF_TENANT_ID
-            ]
+            WorkspaceConfig(workspace)[WorkspaceConfig.CONF_WS_ID]: WorkspaceConfig(
+                workspace
+            )[WorkspaceConfig.CONF_TENANT_ID]
             for workspace in workspaces
         }
         if len(set(workspace_configs.values())) > 1:
@@ -681,7 +710,9 @@ def _schema_format_tables(
 
 def _schema_format_columns(table_schema: dict[str, Any]) -> dict[str, str]:
     """Return a sorted dictionary of column names and types."""
-    columns = {col["name"]: col["type"] for col in table_schema.get("standardColumns", {})}
+    columns = {
+        col["name"]: col["type"] for col in table_schema.get("standardColumns", {})
+    }
     for col in table_schema.get("customColumns", []):
         columns[col["name"]] = col["type"]
     return dict(sorted(columns.items()))

tests/data/drivers/test_azure_monitor_driver.py

@@ -225,6 +225,20 @@ def test_get_schema(az_connect, read_schema, monkeypatch):
     check.is_false(azmon_driver.schema)
 
 
+def test_workspace_args_are_normalized(monkeypatch):
+    """Test that workspace Args are mapped to az_connect keyword names."""
+    with custom_mp_config(
+        get_test_data_path().parent.joinpath("msticpyconfig-test.yaml")
+    ):
+        azmon_driver = AzureMonitorDriver(debug=True)
+        azmon_driver._get_workspaces(workspace="MyTestWS")
+        connect_kwargs = azmon_driver._get_workspace_settings_args()
+
+    check.equal(connect_kwargs["client_id"], "11111111-1111-1111-1111-111111111111")
+    check.equal(connect_kwargs["certificate_path"], "c:/certs/test-cert.pem")
+    check.equal(connect_kwargs["password"], "[PLACEHOLDER]")
+
+
 def test_query_not_connected():
     """Test KqlDriverAZMon query when not connected."""
     with pytest.raises(MsticpyNotConnectedError):

tests/msticpyconfig-test.yaml

@@ -16,6 +16,10 @@ AzureSentinel:
       SubscriptionId: "cd928da3-dcde-42a3-aad7-d2a1268c2f48"
       ResourceGroup: ABC
       WorkspaceName: Workspace1
+      Args:
+        ClientId: "11111111-1111-1111-1111-111111111111"
+        Certificate: "c:/certs/test-cert.pem"
+        CertificatePassword: "[PLACEHOLDER]"
     MyTestWS2:
       WorkspaceId: "a927809c-8142-43e1-96b3-4ad87cfe95a4"
       TenantId: "69d28fd7-42a5-48bc-a619-af56397b9f28"