refactor: Add user drift detection to azure cli auth mode

src/fabric_cli/commands/auth/fab_auth.py

@@ -220,8 +220,7 @@ def logout(args: Namespace) -> None:
 
 def status(args: Namespace) -> None:
     auth = FabAuth()
-    identity_type = auth.get_identity_type()
-    tenant_id = auth.get_tenant_id()
+    initial_identity_type = auth.get_identity_type()
 
     def __get_token_info(scope):
         try:
@@ -240,11 +239,6 @@ def __get_token_info(scope):
 
     token_info = __get_token_info(fab_constant.SCOPE_FABRIC_DEFAULT)
 
-    upn = token_info.get("upn") or "N/A"
-    oid = token_info.get("oid") or "N/A"
-    tid = token_info.get("tid", tenant_id) or "N/A"
-    appid = token_info.get("appid") or "N/A"
-
     def __mask_token(scope):
         try:
             token = auth.get_access_token(scope, interactive_renew=False)
@@ -268,6 +262,18 @@ def __mask_token(scope):
     storage_secret = __mask_token(fab_constant.SCOPE_ONELAKE_DEFAULT)
     azure_secret = __mask_token(fab_constant.SCOPE_AZURE_DEFAULT)
 
+    identity_type = auth.get_identity_type()
+    tenant_id = auth.get_tenant_id()
+
+    if identity_type is None and initial_identity_type == "azure_cli":
+        token_info = {}
+        fabric_secret, storage_secret, azure_secret = ("N/A",) * 3
+
+    upn = token_info.get("upn") or "N/A"
+    oid = token_info.get("oid") or "N/A"
+    tid = token_info.get("tid", tenant_id) or "N/A"
+    appid = token_info.get("appid") or "N/A"
+
     # Check login status
     is_logged_in = fabric_secret != "N/A"
     login_status = (

src/fabric_cli/core/fab_auth.py

@@ -328,6 +328,11 @@ def get_tenant_id(self):
     def get_identity_type(self):
         return self._get_auth_property(con.IDENTITY_TYPE)
 
+    def validate_azure_cli_identity(self) -> None:
+        """Validate the current Azure CLI identity when that mode is active."""
+        if self.get_identity_type() == "azure_cli":
+            self.get_access_token(con.SCOPE_FABRIC_DEFAULT, interactive_renew=False)
+
     def set_access_mode(self, mode, tenant_id=None):
         if mode not in con.AUTH_KEYS[con.IDENTITY_TYPE]:
             raise FabricCLIError(
@@ -338,6 +343,10 @@ def set_access_mode(self, mode, tenant_id=None):
             )
         if mode != self.get_identity_type():
             self.logout()
+        elif mode == "azure_cli":
+            # Reset the baseline so an explicit re-login establishes the
+            # current Azure CLI identity instead of reporting identity drift
+            self._reset_azure_cli_identity_baseline()
         if tenant_id and self.get_tenant_id() != tenant_id:
             self.set_tenant(tenant_id)
         self._set_auth_property(con.IDENTITY_TYPE, mode)
@@ -421,10 +430,10 @@ def set_managed_identity(self, client_id=None):
         )
 
     def _acquire_token_from_azure_cli(self, scope: list[str]) -> dict:
-        """Acquire a token using the current Azure CLI authentication context.
+        """Acquire a token and validate its identity against the stored baseline.
 
-        Synchronizes Fabric CLI's tenant, resource caches, and command context
-        with the tenant claim in the acquired token.
+        Records the tenant and principal on first use. If either identity later
+        changes, logs out and clears cached Fabric CLI state.
         """
         from azure.core.exceptions import (
             ClientAuthenticationError,
@@ -440,11 +449,9 @@ def _acquire_token_from_azure_cli(self, scope: list[str]) -> dict:
             # AzureCliCredential.get_token expects scopes as positional args
             azure_token = self._azure_cli_credential.get_token(scope[0])
 
-            # Keep tenant-scoped context and caches aligned with Azure CLI
+            # Extract claims from the acquired token to determine the current Azure CLI identity
             claims = self._decode_jwt_token(azure_token.token)
-            tid = claims.get("tid")
-            if tid and tid != self.get_tenant_id():
-                self._synchronize_azure_cli_tenant(tid)
+            self._check_azure_cli_identity(claims)
 
             token_result = {
                 "access_token": azure_token.token,
@@ -478,14 +485,68 @@ def _acquire_token_from_azure_cli(self, scope: list[str]) -> dict:
                 status_code=con.ERROR_AUTHENTICATION_FAILED,
             )
 
-    def _synchronize_azure_cli_tenant(self, tenant_id: str) -> None:
-        """Synchronize state when the active Azure CLI tenant changes."""
+    def _check_azure_cli_identity(self, claims: dict) -> None:
+        """Records Azure CLI tenant and principal IDs and rejects identity drift."""
         from fabric_cli.core.fab_context import Context
         from fabric_cli.utils import fab_mem_store
 
-        self._set_auth_properties({con.FAB_TENANT_ID: tenant_id})
-        fab_mem_store.clear_caches()
-        Context().context = self.get_tenant()
+        # Get the tenant and principal IDs from the claims
+        tenant_id = claims.get("tid")
+        principal_id = claims.get("oid")
+
+        if tenant_id is None or principal_id is None:
+            raise FabricCLIError(
+                ErrorMessages.Auth.azure_cli_identity_claims_missing(),
+                status_code=con.ERROR_AUTHENTICATION_FAILED,
+            )
+
+        # Get the current tenant and principal IDs from the auth context
+        current_tenant_id = self.get_tenant_id()
+        current_principal_id = self._get_auth_property(con.FAB_PRINCIPAL_ID)
+
+        # Determine if there is an identity drift (tenant or principal)
+        tenant_drifted = (
+            current_tenant_id is not None and tenant_id != current_tenant_id
+        )
+        principal_drifted = (
+            current_principal_id is not None and principal_id != current_principal_id
+        )
+        # If any drift is detected, set the changed identity description
+        if tenant_drifted or principal_drifted:
+            if tenant_drifted and principal_drifted:
+                changed_identity = "Tenant ID and Principal ID"
+            elif tenant_drifted:
+                changed_identity = "Tenant ID"
+            else:
+                changed_identity = "Principal ID"
+            fab_logger.log_warning(f"Change detected in Azure CLI {changed_identity}")
+
+            # Logout, clear caches, and reset context before raising an error
+            self.logout()
+            fab_mem_store.clear_caches()
+            Context().reset_context()
+            raise FabricCLIError(
+                ErrorMessages.Auth.azure_cli_identity_changed(),
+                status_code=con.ERROR_AUTHENTICATION_FAILED,
+            )
+
+        # Record the initial Azure CLI identity as the baseline for future drift checks
+        auth_properties: dict[str, str] = {}
+
+        if current_tenant_id is None:
+            auth_properties[con.FAB_TENANT_ID] = tenant_id
+        if current_principal_id is None:
+            auth_properties[con.FAB_PRINCIPAL_ID] = principal_id
+        if auth_properties:
+            self._set_auth_properties(auth_properties)
+        if current_tenant_id is None:
+            Context().context = self.get_tenant()
+
+    def _reset_azure_cli_identity_baseline(self):
+        self._auth_info.pop(con.FAB_TENANT_ID, None)
+        self._auth_info.pop(con.FAB_PRINCIPAL_ID, None)
+        self._azure_cli_credential = None
+        self._save_auth()
 
     def print_auth_info(self):
         utils_ui.print_grey(json.dumps(self._get_auth_info(), indent=2))

src/fabric_cli/core/fab_constant.py

@@ -54,6 +54,7 @@
 FAB_SPN_CERT_PASSWORD = "fab_spn_cert_password"
 FAB_SPN_FEDERATED_TOKEN = "fab_spn_federated_token"
 FAB_TENANT_ID = "fab_tenant_id"
+FAB_PRINCIPAL_ID = "fab_principal_id"
 
 FAB_REFRESH_TOKEN = "fab_refresh_token"
 IDENTITY_TYPE = "identity_type"

src/fabric_cli/core/fab_decorators.py

@@ -8,6 +8,7 @@
     ERROR_UNAUTHORIZED,
     EXIT_CODE_AUTHORIZATION_REQUIRED,
     EXIT_CODE_ERROR,
+    FAB_MODE_INTERACTIVE,
 )
 from fabric_cli.core.fab_exceptions import FabricCLIError
 from fabric_cli.utils import fab_ui
@@ -21,7 +22,7 @@ def getinstance(*args, **kwargs):
         if class_ not in instances:
             instances[class_] = class_(*args, **kwargs)
         return instances[class_]
-    
+
     return getinstance
 
 
@@ -64,8 +65,14 @@ def decorator(func):
         def wrapper(*args, **kwargs):
             # Import Context locally to avoid circular import
             from fabric_cli.core.fab_context import Context
-            Context().command = args[0].command_path
-            Context().fabric_skill = getattr(args[0], "skill", None)
+
+            context = Context()
+            context.command = args[0].command_path
+            context.fabric_skill = getattr(args[0], "skill", None)
+            if context.get_runtime_mode() == FAB_MODE_INTERACTIVE:
+                from fabric_cli.core.fab_auth import FabAuth
+
+                FabAuth().validate_azure_cli_identity()
             return func(*args, **kwargs)
 
         return wrapper

src/fabric_cli/errors/auth.py

@@ -138,6 +138,17 @@ def azure_cli_token_acquisition_failed() -> str:
             "Run 'az login' to authenticate, then retry"
         )
 
+    @staticmethod
+    def azure_cli_identity_claims_missing() -> str:
+        return "Azure CLI token is missing required tenant or principal identity claims"
+
+    @staticmethod
+    def azure_cli_identity_changed() -> str:
+        return (
+            "Fabric CLI logged out due to change in Azure CLI identity. "
+            "Run `fab auth login --azure-cli` to re-authenticate with the current Azure CLI identity"
+        )
+
     @staticmethod
     def incompatible_authentication_arguments(arguments: list[str]) -> str:
         return f"Authentication arguments cannot be combined: {', '.join(arguments)}"

tests/conftest.py

@@ -118,7 +118,11 @@ def azure_cli_auth_fixture(monkeypatch, tmp_path):
     auth = FabAuth()
     monkeypatch.setattr(auth, "auth_file", str(tmp_path / "auth.json"))
     monkeypatch.setattr(auth, "cache_file", str(tmp_path / "cache.bin"))
-    monkeypatch.setattr(auth, "_decode_jwt_token", lambda _: {"tid": "test-tenant"})
+    monkeypatch.setattr(
+        auth,
+        "_decode_jwt_token",
+        lambda _: {"tid": "test-tenant", "oid": "test-principal"},
+    )
     auth._azure_cli_credential = None
     auth._auth_info = {}
     auth.app = None

tests/test_commands/test_auth.py

@@ -881,9 +881,13 @@ def test_auth_logout(
         args = argparse.Namespace()
 
         # Act
-        fab_auth.logout(args)
+        with patch.object(
+            fab_auth.utils_mem_store, "clear_caches"
+        ) as mock_clear_caches:
+            fab_auth.logout(args)
 
         # Assert
+        mock_clear_caches.assert_called_once_with()
         mock_fab_context_instance = mock_fab_context.get("instance")
         mock_fab_context_instance.reset_context.assert_called_once()
 
@@ -919,14 +923,18 @@ def test_auth_status(self, mock_fab_auth, capsys):
             auth_subcommand="status",
             output_format="text",
         )
-        with patch(
-            "fabric_cli.commands.auth.fab_auth._get_token_info_from_bearer_token",
-            return_value={
-                "appid": "mocked_appid",
-                "upn": "mocked_upn",
-                "oid": "mocked_oid",
-                "tid": "mocked_tenant_id",
-            },
+        auth = mock_fab_auth["instance"]
+        with (
+            patch.object(auth, "get_identity_type", return_value="user"),
+            patch(
+                "fabric_cli.commands.auth.fab_auth._get_token_info_from_bearer_token",
+                return_value={
+                    "appid": "mocked_appid",
+                    "upn": "mocked_upn",
+                    "oid": "mocked_oid",
+                    "tid": "mocked_tenant_id",
+                },
+            ),
         ):
             # Act
             fab_auth.status(args)
@@ -989,6 +997,96 @@ def test_auth_status_azure_cli_session_unavailable(self, mock_fab_auth, capsys):
         assert "Azure CLI Session: Unavailable" in captured.out
         assert "Logged In: False" in captured.out
 
+    def test_auth_status_identity_drift_during_token_masking(
+        self, mock_fab_auth, capsys
+    ):
+        """Status should report logged-out state after Azure CLI identity drift."""
+        args = argparse.Namespace(
+            command="auth",
+            auth_subcommand="status",
+            output_format="text",
+        )
+        auth = mock_fab_auth["instance"]
+        auth._auth_info = {
+            fab_constant.IDENTITY_TYPE: "azure_cli",
+            fab_constant.FAB_TENANT_ID: "previous-tenant",
+        }
+        auth.get_tenant_id.side_effect = lambda: auth._auth_info.get(
+            fab_constant.FAB_TENANT_ID
+        )
+
+        def get_access_token(*args, **kwargs):
+            if auth.get_access_token.call_count <= 2:
+                return "mocked_access_token"
+            auth._auth_info = {}
+            raise FabricCLIError(
+                ErrorMessages.Auth.azure_cli_identity_changed(),
+                fab_constant.ERROR_AUTHENTICATION_FAILED,
+            )
+
+        auth.get_access_token.side_effect = get_access_token
+
+        with patch(
+            "fabric_cli.commands.auth.fab_auth._get_token_info_from_bearer_token",
+            return_value={"tid": "previous-tenant"},
+        ):
+            fab_auth.status(args)
+
+        captured = capsys.readouterr()
+        assert "Not logged in to app.fabric.microsoft.com" in captured.err
+        assert "Authentication Mode: Azure CLI" not in captured.out
+        assert "Azure CLI Session:" not in captured.out
+        assert "Account: N/A" in captured.out
+        assert "Principal ID: N/A" in captured.out
+        assert "Tenant ID: N/A" in captured.out
+        assert "App ID: N/A" in captured.out
+        assert "Token Fabric PowerBI: N/A" in captured.out
+        assert "Token Storage: N/A" in captured.out
+        assert "Token Azure: N/A" in captured.out
+        assert "Logged In: False" in captured.out
+        assert "previous-tenant" not in captured.out
+
+    def test_auth_status_without_identity_preserves_available_tokens(
+        self, mock_fab_auth, capsys
+    ):
+        args = argparse.Namespace(
+            command="auth",
+            auth_subcommand="status",
+            output_format="text",
+        )
+        auth = mock_fab_auth["instance"]
+
+        with (
+            patch.object(auth, "get_identity_type", return_value=None),
+            patch(
+                "fabric_cli.commands.auth.fab_auth._get_token_info_from_bearer_token",
+                return_value={},
+            ),
+        ):
+            fab_auth.status(args)
+
+        captured = capsys.readouterr()
+        assert "Logged in to app.fabric.microsoft.com" in captured.err
+        assert "Logged In: True" in captured.out
+        assert (
+            "Token Fabric PowerBI: mock************************************"
+            in captured.out
+        )
+        assert "Token Storage: mock************************************" in captured.out
+        assert "Token Azure: mock************************************" in captured.out
+        assert [
+            token_call.args[0] for token_call in auth.get_access_token.call_args_list
+        ] == [
+            fab_constant.SCOPE_FABRIC_DEFAULT,
+            fab_constant.SCOPE_FABRIC_DEFAULT,
+            fab_constant.SCOPE_ONELAKE_DEFAULT,
+            fab_constant.SCOPE_AZURE_DEFAULT,
+        ]
+        assert all(
+            token_call.kwargs == {"interactive_renew": False}
+            for token_call in auth.get_access_token.call_args_list
+        )
+
     def test_init_when_user_cancels_the_prompt(
         self, mock_fab_auth, mock_fab_context, mock_fab_logger_log_warning, capsys
     ):