This is an automated email from the ASF dual-hosted git repository.
dabla pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 198944a3ce3 Support certificate auth for Microsoft Graph filesystem
(#71362)
198944a3ce3 is described below
commit 198944a3ce3753c7169296c3c1b7f37d175de141
Author: Vincent Hsiao <[email protected]>
AuthorDate: Tue Aug 11 00:49:25 2026 +0800
Support certificate auth for Microsoft Graph filesystem (#71362)
---
providers/microsoft/azure/docs/changelog.rst | 1 +
.../microsoft/azure/docs/filesystems/msgraph.rst | 8 +-
.../providers/microsoft/azure/fs/msgraph.py | 64 ++++++++++++-
.../tests/unit/microsoft/azure/fs/test_msgraph.py | 106 ++++++++++++++++++++-
4 files changed, 170 insertions(+), 9 deletions(-)
diff --git a/providers/microsoft/azure/docs/changelog.rst
b/providers/microsoft/azure/docs/changelog.rst
index 523924e7e9b..bed6667dca8 100644
--- a/providers/microsoft/azure/docs/changelog.rst
+++ b/providers/microsoft/azure/docs/changelog.rst
@@ -33,6 +33,7 @@ Changelog
Features
~~~~~~~~
+* ``Add certificate-based authentication support to Microsoft Graph filesystem
(#69335)``
* ``Add WasbRemoteLogIO.from_config and register wasb remote logging scheme
(#70301)``
Bug Fixes
diff --git a/providers/microsoft/azure/docs/filesystems/msgraph.rst
b/providers/microsoft/azure/docs/filesystems/msgraph.rst
index 7ebea783b7e..90d8c5e40a7 100644
--- a/providers/microsoft/azure/docs/filesystems/msgraph.rst
+++ b/providers/microsoft/azure/docs/filesystems/msgraph.rst
@@ -35,13 +35,15 @@ Create a Microsoft Graph connection in Airflow with the
following parameters:
* **Connection Type**: msgraph
* **Host**: Tenant ID
* **Login**: Client ID
-* **Password**: Client Secret
+* **Password**: Client Secret, or certificate password when using
certificate-based authentication
The connection form provides additional configuration fields:
* **Tenant ID**: Azure AD tenant identifier
* **Drive ID**: Specific drive to access (optional - leave empty for general
access)
* **Scopes**: OAuth2 scopes (default: https://graph.microsoft.com/.default)
+* **Certificate path**: File path to a PEM certificate for certificate-based
authentication
+* **Certificate data**: PEM certificate data for certificate-based
authentication
Additional OAuth2 parameters supported via connection extras:
@@ -52,6 +54,10 @@ Additional OAuth2 parameters supported via connection extras:
* **code_challenge_method**: PKCE code challenge method (e.g., 'S256')
* **username**: Username for password grant flow
* **password**: Password for password grant flow
+* **certificate_path**: File path to a PEM certificate
+* **certificate_data**: PEM certificate data
+* **authority**: Microsoft Entra authority host
+* **disable_instance_discovery**: Disable Microsoft Entra instance discovery
Connection extra field configuration example:
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/fs/msgraph.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/fs/msgraph.py
index f0988f961b6..b6fb11f76a3 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/fs/msgraph.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/fs/msgraph.py
@@ -26,6 +26,42 @@ if TYPE_CHECKING:
from fsspec import AbstractFileSystem
schemes = ["msgraph", "sharepoint", "onedrive", "msgd"]
+DEFAULT_SCOPE = "https://graph.microsoft.com/.default";
+
+
+def _get_token_endpoint(tenant_id: str) -> str:
+ return f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token";
+
+
+def _get_scopes(options: dict[str, Any]) -> list[str]:
+ scopes = options.get("scope") or options.get("scopes") or DEFAULT_SCOPE
+ if isinstance(scopes, str):
+ return scopes.split()
+ return scopes
+
+
+def _get_certificate_token(options: dict[str, Any]) -> dict[str, Any]:
+ from azure.identity import CertificateCredential
+
+ credential = CertificateCredential(
+ tenant_id=options["tenant_id"],
+ client_id=options["client_id"],
+ password=options.get("password") or options.get("client_secret"),
+ certificate_path=options.get("certificate_path"),
+ certificate_data=options["certificate_data"].encode() if
options.get("certificate_data") else None,
+ authority=options.get("authority"),
+ disable_instance_discovery=options.get("disable_instance_discovery",
False),
+ )
+ try:
+ access_token = credential.get_token(*_get_scopes(options))
+ finally:
+ credential.close()
+
+ return {
+ "access_token": access_token.token,
+ "token_type": "Bearer",
+ "expires_at": access_token.expires_on,
+ }
def get_fs(conn_id: str | None, storage_options: dict[str, Any] | None = None)
-> AbstractFileSystem:
@@ -62,6 +98,7 @@ def get_fs(conn_id: str | None, storage_options: dict[str,
Any] | None = None) -
fields = [
"drive_id",
"scope",
+ "scopes",
"token_endpoint",
"redirect_uri",
"token_endpoint_auth_method",
@@ -69,6 +106,10 @@ def get_fs(conn_id: str | None, storage_options: dict[str,
Any] | None = None) -
"update_token",
"username",
"password",
+ "certificate_path",
+ "certificate_data",
+ "authority",
+ "disable_instance_discovery",
]
for field in fields:
value = get_field(conn_id=conn_id, conn_type=conn_type, extras=extras,
field_name=field)
@@ -83,7 +124,19 @@ def get_fs(conn_id: str | None, storage_options: dict[str,
Any] | None = None) -
# Create oauth2 client parameters if authentication is provided
oauth2_client_params = {}
- if options.get("client_id") and options.get("client_secret") and
options.get("tenant_id"):
+ if (
+ options.get("client_id")
+ and options.get("tenant_id")
+ and (options.get("certificate_path") or
options.get("certificate_data"))
+ ):
+ token_endpoint = options.get("token_endpoint") or
_get_token_endpoint(options["tenant_id"])
+ oauth2_client_params = {
+ "client_id": options["client_id"],
+ "token": _get_certificate_token(options),
+ "token_endpoint": token_endpoint,
+ "scope": " ".join(_get_scopes(options)),
+ }
+ elif options.get("client_id") and options.get("client_secret") and
options.get("tenant_id"):
oauth2_client_params = {
"client_id": options["client_id"],
"client_secret": options["client_secret"],
@@ -105,11 +158,12 @@ def get_fs(conn_id: str | None, storage_options:
dict[str, Any] | None = None) -
if param in options:
oauth2_client_params[param] = options[param]
+ if "scopes" in options and "scope" not in oauth2_client_params:
+ oauth2_client_params["scope"] = " ".join(_get_scopes(options))
+
# Construct default token_endpoint from tenant_id if not explicitly
provided
- if "token_endpoint" not in oauth2_client_params and tenant_id:
- oauth2_client_params["token_endpoint"] = (
-
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token";
- )
+ if "token_endpoint" not in oauth2_client_params:
+ oauth2_client_params["token_endpoint"] =
_get_token_endpoint(options["tenant_id"])
# Determine which filesystem to return based on drive_id
drive_id = options.get("drive_id")
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/fs/test_msgraph.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/fs/test_msgraph.py
index d95a772dda2..92a21422752 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/fs/test_msgraph.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/fs/test_msgraph.py
@@ -137,14 +137,18 @@ class TestMSGraphFS:
mock_fs_instance = MagicMock()
mock_msgdrivefs.return_value = mock_fs_instance
- storage_options = {"drive_id": "storage_drive_id", "scope":
"custom.scope"}
+ storage_options = {
+ "drive_id": "storage_drive_id",
+ "scope": "custom.scope",
+ "tenant_id": "storage_tenant_id",
+ }
result = get_fs("msgraph_minimal", storage_options=storage_options)
expected_oauth2_params = {
"client_id": "test_client_id",
"client_secret": "test_client_secret",
- "tenant_id": "test_tenant_id",
- "token_endpoint":
"https://login.microsoftonline.com/test_tenant_id/oauth2/v2.0/token";,
+ "tenant_id": "storage_tenant_id",
+ "token_endpoint":
"https://login.microsoftonline.com/storage_tenant_id/oauth2/v2.0/token";,
"scope": "custom.scope",
}
mock_msgdrivefs.assert_called_once_with(
@@ -152,6 +156,102 @@ class TestMSGraphFS:
)
assert result == mock_fs_instance
+ @patch("azure.identity.CertificateCredential", autospec=True)
+
@patch("airflow.providers.microsoft.azure.fs.msgraph.BaseHook.get_connection",
autospec=True)
+ @patch("msgraphfs.MSGDriveFS", autospec=True)
+ def test_get_fs_with_certificate_path(
+ self, mock_msgdrivefs, mock_get_connection, mock_certificate_credential
+ ):
+ connection = Connection(
+ conn_id="msgraph_certificate",
+ conn_type="msgraph",
+ login="test_client_id",
+ password="certificate_password",
+ host="test_tenant_id",
+ extra={"drive_id": "test_drive_id", "certificate_path":
"/tmp/cert.pem"},
+ )
+ mock_get_connection.return_value = connection
+ mock_fs_instance = MagicMock()
+ mock_msgdrivefs.return_value = mock_fs_instance
+ mock_access_token = MagicMock(token="certificate-token",
expires_on=1234567890)
+ mock_certificate_credential.return_value.get_token.return_value =
mock_access_token
+
+ result = get_fs("msgraph_certificate")
+
+ mock_certificate_credential.assert_called_once_with(
+ tenant_id="test_tenant_id",
+ client_id="test_client_id",
+ password="certificate_password",
+ certificate_path="/tmp/cert.pem",
+ certificate_data=None,
+ authority=None,
+ disable_instance_discovery=False,
+ )
+
mock_certificate_credential.return_value.get_token.assert_called_once_with(
+ "https://graph.microsoft.com/.default";
+ )
+
mock_certificate_credential.return_value.close.assert_called_once_with()
+ mock_msgdrivefs.assert_called_once_with(
+ drive_id="test_drive_id",
+ oauth2_client_params={
+ "client_id": "test_client_id",
+ "token": {
+ "access_token": "certificate-token",
+ "token_type": "Bearer",
+ "expires_at": 1234567890,
+ },
+ "token_endpoint":
"https://login.microsoftonline.com/test_tenant_id/oauth2/v2.0/token";,
+ "scope": "https://graph.microsoft.com/.default";,
+ },
+ )
+ assert result == mock_fs_instance
+
+ @patch("azure.identity.CertificateCredential", autospec=True)
+
@patch("airflow.providers.microsoft.azure.fs.msgraph.BaseHook.get_connection",
autospec=True)
+ @patch("msgraphfs.MSGDriveFS", autospec=True)
+ def test_get_fs_with_certificate_data_from_storage_options(
+ self, mock_msgdrivefs, mock_get_connection,
mock_certificate_credential, mock_connection_minimal
+ ):
+ mock_get_connection.return_value = mock_connection_minimal
+ mock_fs_instance = MagicMock()
+ mock_msgdrivefs.return_value = mock_fs_instance
+ mock_access_token = MagicMock(token="storage-token",
expires_on=1234567890)
+ mock_certificate_credential.return_value.get_token.return_value =
mock_access_token
+
+ result = get_fs(
+ "msgraph_minimal",
+ storage_options={
+ "certificate_data": "certificate-data",
+ "scope": "custom.scope",
+ "token_endpoint":
"https://login.microsoftonline.com/custom/oauth2/v2.0/token";,
+ },
+ )
+
+ mock_certificate_credential.assert_called_once_with(
+ tenant_id="test_tenant_id",
+ client_id="test_client_id",
+ password="test_client_secret",
+ certificate_path=None,
+ certificate_data=b"certificate-data",
+ authority=None,
+ disable_instance_discovery=False,
+ )
+
mock_certificate_credential.return_value.get_token.assert_called_once_with("custom.scope")
+ mock_msgdrivefs.assert_called_once_with(
+ drive_id=None,
+ oauth2_client_params={
+ "client_id": "test_client_id",
+ "token": {
+ "access_token": "storage-token",
+ "token_type": "Bearer",
+ "expires_at": 1234567890,
+ },
+ "token_endpoint":
"https://login.microsoftonline.com/custom/oauth2/v2.0/token";,
+ "scope": "custom.scope",
+ },
+ )
+ assert result == mock_fs_instance
+
@patch("airflow.providers.microsoft.azure.fs.msgraph.BaseHook.get_connection")
@patch("msgraphfs.MSGDriveFS")
def test_get_fs_incomplete_credentials(self, mock_msgdrivefs,
mock_get_connection):
