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,21 @@ 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()
+
+    # Reacquire status after drift to discard stale values and allow env token fallback
+    if identity_type is None and initial_identity_type == "azure_cli":
+        token_info = __get_token_info(fab_constant.SCOPE_FABRIC_DEFAULT)
+        fabric_secret = __mask_token(fab_constant.SCOPE_FABRIC_DEFAULT)
+        storage_secret = __mask_token(fab_constant.SCOPE_ONELAKE_DEFAULT)
+        azure_secret = __mask_token(fab_constant.SCOPE_AZURE_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"
+
     # 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":
+            # Clear the baseline so an explicit re-login establishes the
+            # current Azure CLI identity instead of reporting identity drift
+            self._clear_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 an error to stop the current operation upon detecting identity drift
+            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)
+            Context().context = self.get_tenant()
+
+    def _clear_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 returned an invalid token. Run 'az login' to authenticate, then retry"
+
+    @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