feat: add issuer (iss) to drift detection key for sovereign cloud safety

- Add FAB_AZURE_CLI_ISSUER constant and store iss claim at login
- Check iss before tid and oid on every token acquisition
- Drift key is now iss + tid + oid (environment + tenant + identity)
- Add azure_cli_environment_mismatch error message
- Add TestAzureCliEnvironmentDrift tests (public vs sovereign)
- Update JWT test helpers to include iss claim

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: Shira Sassoon

src/fabric_cli/core/fab_auth.py

@@ -455,16 +455,18 @@ def set_azure_cli(self, tenant_id=None):
 
         # Set identity_type after tenant to survive any logout triggered by tenant change
         auth_props: dict = {con.IDENTITY_TYPE: "azure_cli"}
-        # Store OID for drift detection (immutable, no PII)
+        # Store OID and issuer for drift detection (immutable, no PII)
         if claims.get("oid"):
             auth_props[con.FAB_AZURE_CLI_PRINCIPAL_ID] = claims["oid"]
+        if claims.get("iss"):
+            auth_props[con.FAB_AZURE_CLI_ISSUER] = claims["iss"]
         self._set_auth_properties(auth_props)
 
     @staticmethod
     def _decode_jwt_claims(token: str) -> dict:
         """Decode JWT payload claims without signature validation.
 
-        Used to extract identity claims (tid, oid) from tokens
+        Used to extract identity claims (iss, tid, oid) from tokens
         returned by AzureCliCredential. Signature validation is
         unnecessary here — the token was just returned by the
         Azure CLI SDK over a local subprocess call.
@@ -485,8 +487,8 @@ def _acquire_token_from_azure_cli(self, scope: list[str]) -> dict:
         """Acquire a token using Azure CLI's AzureCliCredential.
 
         After acquiring the token, decodes JWT claims and verifies
-        that tid and oid match the stored values from login to detect
-        identity drift (e.g., user ran 'az login' as a different user).
+        that iss, tid, and oid match the stored values from login to detect
+        identity or environment drift.
         """
         stored_tenant = self.get_tenant_id()
 
@@ -504,6 +506,15 @@ def _acquire_token_from_azure_cli(self, scope: list[str]) -> dict:
             # Post-acquisition drift detection from actual token claims
             claims = self._decode_jwt_claims(azure_token.token)
 
+            # Environment drift check (issuer encodes cloud: public vs sovereign)
+            stored_issuer = self._auth_info.get(con.FAB_AZURE_CLI_ISSUER)
+            if stored_issuer and claims.get("iss"):
+                if claims["iss"] != stored_issuer:
+                    raise FabricCLIError(
+                        ErrorMessages.Auth.azure_cli_environment_mismatch(),
+                        status_code=con.ERROR_AUTHENTICATION_FAILED,
+                    )
+
             # Tenant drift check
             if stored_tenant and claims.get("tid"):
                 if claims["tid"] != stored_tenant:

src/fabric_cli/core/fab_constant.py

@@ -57,6 +57,7 @@
 
 FAB_REFRESH_TOKEN = "fab_refresh_token"
 FAB_AZURE_CLI_PRINCIPAL_ID = "fab_azure_cli_principal_id"
+FAB_AZURE_CLI_ISSUER = "fab_azure_cli_issuer"
 IDENTITY_TYPE = "identity_type"
 FAB_AUTH_MODE = "fab_auth_mode"  # Kept for backward compatibility
 FAB_AUTHORITY = "fab_authority"

src/fabric_cli/errors/auth.py

@@ -128,6 +128,13 @@ def azure_cli_tenant_mismatch(stored_tenant: str, current_tenant: str) -> str:
             "Run 'fab auth login --azure-cli' to re-authenticate."
         )
 
+    @staticmethod
+    def azure_cli_environment_mismatch() -> str:
+        return (
+            "Azure CLI cloud environment has changed since 'fab auth login --azure-cli' was run. "
+            "Run 'fab auth login --azure-cli' to re-authenticate in the current environment."
+        )
+
     @staticmethod
     def azure_cli_principal_mismatch() -> str:
         return (

tests/test_core/test_fab_auth_azure_cli.py

@@ -14,17 +14,19 @@
 from fabric_cli.errors import ErrorMessages
 
 
-def _make_jwt(tid: str = "test-tenant", oid: str = "test-oid", **extra_claims) -> str:
+def _make_jwt(tid: str = "test-tenant", oid: str = "test-oid",
+              iss: str = "https://sts.windows.net/test-tenant/", **extra_claims) -> str:
     """Create a fake JWT with specified claims (no signature validation needed)."""
     header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode()
-    claims = {"tid": tid, "oid": oid, **extra_claims}
+    claims = {"tid": tid, "oid": oid, "iss": iss, **extra_claims}
     payload = base64.urlsafe_b64encode(_json.dumps(claims).encode()).rstrip(b"=").decode()
     return f"{header}.{payload}.fakesig"
 
 
-def _mock_credential_with_jwt(mock_class, tid="test-tenant", oid="test-oid", **extra):
+def _mock_credential_with_jwt(mock_class, tid="test-tenant", oid="test-oid",
+                               iss="https://sts.windows.net/test-tenant/", **extra):
     """Set up a mock AzureCliCredential that returns a JWT with given claims."""
-    token_str = _make_jwt(tid=tid, oid=oid, **extra)
+    token_str = _make_jwt(tid=tid, oid=oid, iss=iss, **extra)
     mock_token = MagicMock()
     mock_token.token = token_str
     mock_token.expires_on = int(time.time()) + 3600
@@ -280,6 +282,53 @@ def test_tenant_match_allows_token_acquisition(
         assert "access_token" in result
 
 
+class TestAzureCliEnvironmentDrift:
+    """Test cloud environment drift detection via JWT iss claim."""
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_environment_drift_blocks_token_acquisition(
+        self, mock_credential_class, temp_dir_fixture
+    ):
+        """Should block when token issuer differs from stored environment."""
+        # Login in Azure Public
+        _mock_credential_with_jwt(
+            mock_credential_class, tid="t1", oid="u1",
+            iss="https://sts.windows.net/t1/"
+        )
+        auth = FabAuth()
+        auth.set_access_mode("azure_cli")
+        auth.set_azure_cli()
+        auth._azure_cli_credential = None
+
+        # Now credential returns token from Azure Government
+        _mock_credential_with_jwt(
+            mock_credential_class, tid="t1", oid="u1",
+            iss="https://sts.microsoftonline.us/t1/"
+        )
+
+        with pytest.raises(FabricCLIError) as exc_info:
+            auth._acquire_token_from_azure_cli(con.SCOPE_FABRIC_DEFAULT)
+
+        assert "environment has changed" in str(exc_info.value)
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_same_environment_allows_token_acquisition(
+        self, mock_credential_class, temp_dir_fixture
+    ):
+        """Should allow when token issuer matches stored environment."""
+        _mock_credential_with_jwt(
+            mock_credential_class, tid="t1", oid="u1",
+            iss="https://sts.windows.net/t1/"
+        )
+        auth = FabAuth()
+        auth.set_access_mode("azure_cli")
+        auth.set_azure_cli()
+        auth._azure_cli_credential = None
+
+        result = auth._acquire_token_from_azure_cli(con.SCOPE_FABRIC_DEFAULT)
+        assert "access_token" in result
+
+
 class TestAzureCliPrincipalDrift:
     """Test principal (identity) drift detection via JWT OID claims."""
 
@@ -462,14 +511,16 @@ def test_login_discovers_tenant_from_jwt(self, mock_credential_class, temp_dir_f
         assert auth.get_tenant_id() == "discovered-tenant"
 
     @patch("fabric_cli.core.fab_auth.AzureCliCredential")
-    def test_login_stores_oid_for_drift_detection(self, mock_credential_class, temp_dir_fixture):
-        """set_azure_cli should store OID from JWT for drift detection."""
-        _mock_credential_with_jwt(mock_credential_class, tid="t1", oid="user-oid-123")
+    def test_login_stores_oid_and_issuer_for_drift_detection(self, mock_credential_class, temp_dir_fixture):
+        """set_azure_cli should store OID and issuer from JWT for drift detection."""
+        _mock_credential_with_jwt(mock_credential_class, tid="t1", oid="user-oid-123",
+                                   iss="https://sts.windows.net/t1/")
 
         auth = FabAuth()
         auth.set_access_mode("azure_cli")
         auth.set_azure_cli()
         assert auth._auth_info.get(con.FAB_AZURE_CLI_PRINCIPAL_ID) == "user-oid-123"
+        assert auth._auth_info.get(con.FAB_AZURE_CLI_ISSUER) == "https://sts.windows.net/t1/"
 
     @patch("fabric_cli.core.fab_auth.AzureCliCredential")
     def test_re_login_updates_tenant_and_oid(self, mock_credential_class, temp_dir_fixture):

tests/test_core/test_fab_msal_bridge_azure_cli.py

@@ -18,7 +18,7 @@
 def _make_jwt(tid: str = "test-tenant", oid: str = "test-oid") -> str:
     """Create a fake JWT with specified claims."""
     header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode()
-    claims = {"tid": tid, "oid": oid}
+    claims = {"tid": tid, "oid": oid, "iss": f"https://sts.windows.net/{tid}/"}
     payload = base64.urlsafe_b64encode(_json.dumps(claims).encode()).rstrip(b"=").decode()
     return f"{header}.{payload}.fakesig"