Fix Microsoft Graph filesystem auth by defaulting OAuth2 scope

providers/microsoft/azure/src/airflow/providers/microsoft/azure/fs/msgraph.py

@@ -36,7 +36,7 @@ def _get_token_endpoint(tenant_id: str) -> str:
 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 [scope.strip() for scope in scopes.replace(",", " ").split() if scope.strip()]
     return scopes
 
 
@@ -158,7 +158,10 @@ 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:
+        # authlib expects a singular, space-delimited "scope"; the connection form only
+        # offers "scopes", which the hook treats as comma-separated, so translate it and
+        # always default so authlib never authenticates without a scope.
+        if "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

providers/microsoft/azure/tests/unit/microsoft/azure/fs/test_msgraph.py

@@ -64,6 +64,7 @@ def test_get_fs_with_drive_id(self, mock_msgdrivefs, mock_get_connection, mock_c
                 "client_id": "test_client_id",
                 "client_secret": "test_client_secret",
                 "tenant_id": "test_tenant_id",
+                "scope": "https://graph.microsoft.com/.default",
                 "token_endpoint": "https://login.microsoftonline.com/test_tenant_id/oauth2/v2.0/token",
             },
         )
@@ -252,6 +253,70 @@ def test_get_fs_with_certificate_data_from_storage_options(
         )
         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_certificate_rewrites_comma_separated_scopes(
+        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",
+                "scopes": "User.Read,Files.Read",
+            },
+        )
+        mock_get_connection.return_value = connection
+        mock_msgdrivefs.return_value = MagicMock()
+        mock_certificate_credential.return_value.get_token.return_value = MagicMock(
+            token="certificate-token", expires_on=1234567890
+        )
+
+        get_fs("msgraph_certificate")
+
+        mock_certificate_credential.return_value.get_token.assert_called_once_with("User.Read", "Files.Read")
+        assert mock_msgdrivefs.call_args[1]["oauth2_client_params"]["scope"] == "User.Read Files.Read"
+
+    @pytest.mark.parametrize(
+        ("extra", "expected_scope"),
+        [
+            pytest.param({"scope": "explicit.scope"}, "explicit.scope", id="explicit-scope-wins"),
+            pytest.param({"scopes": "form.scope"}, "form.scope", id="falls-back-to-scopes-form-field"),
+            pytest.param(
+                {"scopes": "User.Read,Files.Read"},
+                "User.Read Files.Read",
+                id="rewrites-comma-separated-scopes-to-space-delimited",
+            ),
+            pytest.param(
+                {"scopes": ["User.Read", "Files.Read"]},
+                "User.Read Files.Read",
+                id="joins-list-scopes-into-space-delimited",
+            ),
+            pytest.param({}, "https://graph.microsoft.com/.default", id="defaults-to-graph-scope"),
+        ],
+    )
+    @patch("airflow.providers.microsoft.azure.fs.msgraph.BaseHook.get_connection")
+    @patch("msgraphfs.MSGDriveFS")
+    def test_get_fs_resolves_scope(self, mock_msgdrivefs, mock_get_connection, extra, expected_scope):
+        mock_get_connection.return_value = Connection(
+            conn_id="msgraph_scope",
+            conn_type="msgraph",
+            login="test_client_id",
+            password="test_client_secret",
+            host="test_tenant_id",
+            extra=extra,
+        )
+        mock_msgdrivefs.return_value = MagicMock()
+
+        get_fs("msgraph_scope")
+
+        assert mock_msgdrivefs.call_args[1]["oauth2_client_params"]["scope"] == expected_scope
+
     @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):