refactor: add drift detection caused by env vars

docs/essentials/env_vars.md

@@ -23,4 +23,15 @@ The CLI supports multiple authentication methods through environment variables.
 |  | `FAB_SPN_FEDERATED_TOKEN` | Federated token |
 |  | `FAB_TENANT_ID` | Tenant ID |
 | Managed Identity | `FAB_MANAGED_IDENTITY` | Enable Managed Identity auth (values: `true`, `1`) |
-| | `FAB_SPN_CLIENT_ID` | **Optional**. Service principal client ID for User Assigned |
+| | `FAB_SPN_CLIENT_ID` | **Optional**. Service principal client ID for User Assigned |
+
+## Direct access token identity consistency
+
+When direct access tokens are used, the CLI verifies that all configured token
+variables contain the same tenant (`tid`) and principal (`oid`) claims. The
+token tenant must also match `FAB_TENANT_ID`. If a claim is missing or the
+identities differ, the CLI logs out, clears its authentication and resource
+caches, resets the current context, and fails the command.
+
+This validation applies only to direct access token environment variables. It
+does not apply when Azure CLI authentication mode is active.

src/fabric_cli/commands/auth/fab_auth.py

@@ -209,11 +209,7 @@ def init(args: Namespace) -> Any:
 
 
 def logout(args: Namespace) -> None:
-    FabAuth().logout()
-
-    # Clear cache and context including current and stale context files
-    utils_mem_store.clear_caches()
-    Context().reset_context()
+    FabAuth().logout_session()
 
     fab_ui.print_output_format(args, message="Logged out of Fabric account")
 

src/fabric_cli/core/fab_auth.py

@@ -274,6 +274,7 @@ def _get_app(self) -> msal.ClientApplication:
 
     def _get_access_token_from_env_vars_if_exist(self, scope):
         if "FAB_TOKEN" in os.environ and "FAB_TOKEN_ONELAKE" in os.environ:
+            self._validate_direct_token_identity()
             match scope:
                 case con.SCOPE_FABRIC_DEFAULT:
                     # this call will validate the token we got from the env var
@@ -313,6 +314,43 @@ def _get_access_token_from_env_vars_if_exist(self, scope):
 
         return None
 
+    def _validate_direct_token_identity(self) -> None:
+        token_variables = (
+            ("FAB_TOKEN", con.FABRIC_TOKEN_AUDIENCE),
+            ("FAB_TOKEN_ONELAKE", con.ONELAKE_TOKEN_AUDIENCE),
+            ("FAB_TOKEN_AZURE", con.AZURE_TOKEN_AUDIENCE),
+        )
+        identities = set()
+
+        for variable, audience in token_variables:
+            token = os.environ.get(variable)
+            if token is None:
+                continue
+
+            claims = self._decode_jwt_token(token, audience)
+            tenant_id = claims.get("tid")
+            object_id = claims.get("oid")
+            if not tenant_id or not object_id:
+                self.logout_session()
+                raise FabricCLIError(
+                    ErrorMessages.Auth.direct_token_identity_drift(),
+                    con.ERROR_AUTHENTICATION_FAILED,
+                )
+            identities.add((tenant_id.lower(), object_id.lower()))
+
+        configured_tenant = os.environ.get("FAB_TENANT_ID")
+        token_tenants = {tenant_id for tenant_id, _ in identities}
+        if len(identities) > 1 or (
+            configured_tenant
+            and token_tenants
+            and configured_tenant.lower() not in token_tenants
+        ):
+            self.logout_session()
+            raise FabricCLIError(
+                ErrorMessages.Auth.direct_token_identity_drift(),
+                con.ERROR_AUTHENTICATION_FAILED,
+            )
+
     def get_tenant(self):
         return Tenant(
             name=self.get_tenant_name(),
@@ -526,7 +564,6 @@ def acquire_token(self, scope: list[str], interactive_renew=True) -> dict:
 
         try:
             token = None
-            env_var_token = self._get_access_token_from_env_vars_if_exist(scope)
             identity_type = self.get_identity_type()
 
             if identity_type == "service_principal":
@@ -550,28 +587,30 @@ def acquire_token(self, scope: list[str], interactive_renew=True) -> dict:
                     )
             elif identity_type == "azure_cli":
                 token = self._acquire_token_from_azure_cli(scope)
-            elif env_var_token:
-                token = {
-                    "access_token": env_var_token,
-                }
-            elif identity_type == "user":
-                # Use the cache to get the token
-                accounts = self._get_app().get_accounts()
-                account = None
-                if accounts:
-                    account = accounts[0]
-                token = self._get_app().acquire_token_silent(
-                    scopes=scope, account=account
-                )
-
-                if token is None and interactive_renew:
-                    token = self._get_app().acquire_token_interactive(
-                        scopes=scope,
-                        prompt="select_account",
-                        parent_window_handle=msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE,
+            else:
+                env_var_token = self._get_access_token_from_env_vars_if_exist(scope)
+                if env_var_token:
+                    token = {
+                        "access_token": env_var_token,
+                    }
+                elif identity_type == "user":
+                    # Use the cache to get the token
+                    accounts = self._get_app().get_accounts()
+                    account = None
+                    if accounts:
+                        account = accounts[0]
+                    token = self._get_app().acquire_token_silent(
+                        scopes=scope, account=account
                     )
-                    if token is not None and "id_token_claims" in token:
-                        self.set_tenant(token.get("id_token_claims")["tid"])
+
+                    if token is None and interactive_renew:
+                        token = self._get_app().acquire_token_interactive(
+                            scopes=scope,
+                            prompt="select_account",
+                            parent_window_handle=msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE,
+                        )
+                        if token is not None and "id_token_claims" in token:
+                            self.set_tenant(token.get("id_token_claims")["tid"])
 
             if token and token.get("error"):
                 fab_logger.log_debug(
@@ -654,6 +693,14 @@ def logout(self):
         config.set_config(con.FAB_DEFAULT_AZ_RESOURCE_GROUP, "")
         config.set_config(con.FAB_DEFAULT_AZ_LOCATION, "")
 
+    def logout_session(self) -> None:
+        from fabric_cli.core.fab_context import Context
+        from fabric_cli.utils import fab_mem_store
+
+        self.logout()
+        fab_mem_store.clear_caches()
+        Context().reset_context()
+
     def get_token_claims(
         self, scope: list[str], claim_names: list[str]
     ) -> Optional[dict[str, str]]:

src/fabric_cli/errors/auth.py

@@ -60,6 +60,15 @@ def invalid_scope(scope: str) -> str:
     def both_fab_and_onelake_tokens_required() -> str:
         return "Both FAB_TOKEN and FAB_TOKEN_ONELAKE are required"
 
+    @staticmethod
+    def direct_token_identity_drift() -> str:
+        return (
+            "Direct access token identity drift detected. FAB_TOKEN, "
+            "FAB_TOKEN_ONELAKE, FAB_TOKEN_AZURE, and FAB_TENANT_ID must "
+            "represent the same tenant and principal. The Fabric CLI session "
+            "has been logged out"
+        )
+
     @staticmethod
     def invalid_identity_type(identity_type: str, allowed_values: list) -> str:
         return f"The identity type '{identity_type}' is invalid. Allowed values are: {allowed_values}"

tests/test_core/test_fab_auth.py

@@ -45,6 +45,7 @@ def _clear_environment_variables(monkeypatch):
     monkeypatch.delenv("FAB_SPN_CLIENT_SECRET", raising=False)
     monkeypatch.delenv("FAB_SPN_CERT_PATH", raising=False)
     monkeypatch.delenv("FAB_SPN_CERT_PASSWORD", raising=False)
+    monkeypatch.delenv("FAB_SPN_FEDERATED_TOKEN", raising=False)
     monkeypatch.delenv("FAB_TOKEN", raising=False)
     monkeypatch.delenv("FAB_TOKEN_ONELAKE", raising=False)
     monkeypatch.delenv("FAB_TOKEN_AZURE", raising=False)
@@ -815,6 +816,101 @@ def test_get_access_token_env_var(monkeypatch):
     assert token == "env_token"
 
 
+def test_direct_token_identity_consistent(monkeypatch):
+    _clear_environment_variables(monkeypatch)
+    auth = FabAuth()
+    tenant_id = str(uuid.uuid4())
+    object_id = str(uuid.uuid4())
+    tokens = {
+        "fabric-token": {"tid": tenant_id, "oid": object_id},
+        "onelake-token": {"tid": tenant_id, "oid": object_id},
+        "azure-token": {"tid": tenant_id, "oid": object_id},
+    }
+    monkeypatch.setenv("FAB_TENANT_ID", tenant_id)
+    monkeypatch.setenv("FAB_TOKEN", "fabric-token")
+    monkeypatch.setenv("FAB_TOKEN_ONELAKE", "onelake-token")
+    monkeypatch.setenv("FAB_TOKEN_AZURE", "azure-token")
+    monkeypatch.setattr(
+        auth, "_decode_jwt_token", lambda token, audience: tokens[token]
+    )
+
+    token = auth._get_access_token_from_env_vars_if_exist(con.SCOPE_FABRIC_DEFAULT)
+
+    assert token == "fabric-token"
+
+
+def test_direct_token_identity_drift_logs_out_session(monkeypatch):
+    _clear_environment_variables(monkeypatch)
+    auth = FabAuth()
+    tenant_id = str(uuid.uuid4())
+    tokens = {
+        "fabric-token": {"tid": tenant_id, "oid": str(uuid.uuid4())},
+        "onelake-token": {"tid": tenant_id, "oid": str(uuid.uuid4())},
+        "azure-token": {"tid": tenant_id, "oid": str(uuid.uuid4())},
+    }
+    monkeypatch.setenv("FAB_TENANT_ID", tenant_id)
+    monkeypatch.setenv("FAB_TOKEN", "fabric-token")
+    monkeypatch.setenv("FAB_TOKEN_ONELAKE", "onelake-token")
+    monkeypatch.setenv("FAB_TOKEN_AZURE", "azure-token")
+    monkeypatch.setattr(
+        auth, "_decode_jwt_token", lambda token, audience: tokens[token]
+    )
+
+    with (
+        patch.object(auth, "logout") as mock_logout,
+        patch("fabric_cli.utils.fab_mem_store.clear_caches") as mock_clear_caches,
+        patch("fabric_cli.core.fab_context.Context") as mock_context,
+        pytest.raises(FabricCLIError) as exc_info,
+    ):
+        auth._get_access_token_from_env_vars_if_exist(con.SCOPE_FABRIC_DEFAULT)
+
+    assert exc_info.value.status_code == con.ERROR_AUTHENTICATION_FAILED
+    assert exc_info.value.message == ErrorMessages.Auth.direct_token_identity_drift()
+    mock_logout.assert_called_once_with()
+    mock_clear_caches.assert_called_once_with()
+    mock_context.return_value.reset_context.assert_called_once_with()
+
+
+def test_direct_token_tenant_drift_logs_out_session(monkeypatch):
+    _clear_environment_variables(monkeypatch)
+    auth = FabAuth()
+    token_tenant_id = str(uuid.uuid4())
+    token_claims = {"tid": token_tenant_id, "oid": str(uuid.uuid4())}
+    monkeypatch.setenv("FAB_TENANT_ID", str(uuid.uuid4()))
+    monkeypatch.setenv("FAB_TOKEN", "fabric-token")
+    monkeypatch.setenv("FAB_TOKEN_ONELAKE", "onelake-token")
+    monkeypatch.setattr(auth, "_decode_jwt_token", lambda token, audience: token_claims)
+
+    with (
+        patch.object(auth, "logout_session") as mock_logout_session,
+        pytest.raises(FabricCLIError) as exc_info,
+    ):
+        auth._get_access_token_from_env_vars_if_exist(con.SCOPE_FABRIC_DEFAULT)
+
+    assert exc_info.value.message == ErrorMessages.Auth.direct_token_identity_drift()
+    mock_logout_session.assert_called_once_with()
+
+
+def test_azure_cli_auth_ignores_direct_token_environment(monkeypatch):
+    _clear_environment_variables(monkeypatch)
+    auth = FabAuth()
+    monkeypatch.setattr(auth, "get_identity_type", lambda: "azure_cli")
+    monkeypatch.setattr(
+        auth,
+        "_get_access_token_from_env_vars_if_exist",
+        lambda scope: pytest.fail("Azure CLI auth inspected direct-token variables"),
+    )
+    monkeypatch.setattr(
+        auth,
+        "_acquire_token_from_azure_cli",
+        lambda scope: {"access_token": "azure-cli-token"},
+    )
+
+    token = auth.get_access_token(con.SCOPE_FABRIC_DEFAULT)
+
+    assert token == "azure-cli-token"
+
+
 # -----------------------------
 # User Mode Tests
 # -----------------------------