security: fail-closed on missing JWT identity claims

Reject tokens that lack iss, tid, or oid instead of silently skipping
drift checks. Applied at both login (set_azure_cli) and token
acquisition (_acquire_token_from_azure_cli).

Also fixed redundant nested if in environment drift check and
incorrect indentation in principal drift check.

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

Author: Shira Sassoon

src/fabric_cli/core/fab_auth.py

@@ -448,18 +448,27 @@ def set_azure_cli(self, tenant_id=None):
                 status_code=con.ERROR_AUTHENTICATION_FAILED,
             )
 
+        # Fail-closed: refuse to persist if identity claims are missing
+        if (
+            not claims.get("iss")
+            or not claims.get("tid")
+            or not claims.get("oid")
+        ):
+            raise FabricCLIError(
+                ErrorMessages.Auth.azure_cli_token_missing_claims(),
+                status_code=con.ERROR_AUTHENTICATION_FAILED,
+            )
+
         # Set tenant from explicit param or JWT tid claim
-        resolved_tenant = tenant_id or claims.get("tid")
+        resolved_tenant = tenant_id or claims["tid"]
         if resolved_tenant:
             self.set_tenant(resolved_tenant)
 
         # Set identity_type after tenant to survive any logout triggered by tenant change
         auth_props: dict = {con.IDENTITY_TYPE: "azure_cli"}
         # 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"]
+        auth_props[con.FAB_AZURE_CLI_PRINCIPAL_ID] = claims["oid"]
+        auth_props[con.FAB_AZURE_CLI_ISSUER] = claims["iss"]
         self._set_auth_properties(auth_props)
 
     @staticmethod
@@ -506,30 +515,38 @@ 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)
 
+            # Fail-closed: reject tokens with missing identity claims
+            if (
+                not claims.get("iss")
+                or not claims.get("tid")
+                or not claims.get("oid")
+            ):
+                raise FabricCLIError(
+                    ErrorMessages.Auth.azure_cli_token_missing_claims(),
+                    status_code=con.ERROR_AUTHENTICATION_FAILED,
+                )
+
             # 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(
+            if stored_issuer and 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:
-                    raise FabricCLIError(
-                        ErrorMessages.Auth.azure_cli_tenant_mismatch(
-                            stored_tenant, claims["tid"]
-                        ),
-                        status_code=con.ERROR_AUTHENTICATION_FAILED,
-                    )
+            if stored_tenant and claims["tid"] != stored_tenant:
+                raise FabricCLIError(
+                    ErrorMessages.Auth.azure_cli_tenant_mismatch(
+                        stored_tenant, claims["tid"]
+                    ),
+                    status_code=con.ERROR_AUTHENTICATION_FAILED,
+                )
 
             # Principal drift check (OID-based)
             stored_principal = self._auth_info.get(con.FAB_AZURE_CLI_PRINCIPAL_ID)
-            if stored_principal and claims.get("oid"):
-                if claims["oid"] != stored_principal:
-                    raise FabricCLIError(
+            if stored_principal and claims["oid"] != stored_principal:
+                raise FabricCLIError(
                         ErrorMessages.Auth.azure_cli_principal_mismatch(),
                         status_code=con.ERROR_AUTHENTICATION_FAILED,
                     )

src/fabric_cli/errors/auth.py

@@ -153,6 +153,14 @@ def azure_cli_not_available() -> str:
     def azure_cli_auth_failed(error_msg: str) -> str:
         return f"Azure CLI authentication failed: {error_msg}"
 
+    @staticmethod
+    def azure_cli_token_missing_claims() -> str:
+        return (
+            "Azure CLI returned a token with missing identity claims (iss, tid, or oid). "
+            "Run 'az account get-access-token --resource https://api.fabric.microsoft.com' "
+            "manually to diagnose."
+        )
+
     @staticmethod
     def azure_cli_token_acquisition_failed() -> str:
         return (

tests/test_core/test_fab_auth_azure_cli.py

@@ -599,6 +599,69 @@ def test_jwt_with_extra_claims(self, temp_dir_fixture):
         assert claims["upn"] == "user@contoso.com"
 
 
+class TestFailClosedOnMissingClaims:
+    """Verify tokens with missing identity claims are rejected, not silently used."""
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_login_rejects_token_missing_oid(self, mock_class, temp_dir_fixture):
+        """set_azure_cli should fail if probe token lacks oid."""
+        header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode()
+        payload = base64.urlsafe_b64encode(
+            _json.dumps({"tid": "t1", "iss": "https://sts.windows.net/t1/"}).encode()
+        ).rstrip(b"=").decode()
+        token_str = f"{header}.{payload}.fakesig"
+        mock_token = MagicMock()
+        mock_token.token = token_str
+        mock_token.expires_on = int(time.time()) + 3600
+        mock_class.return_value.get_token.return_value = mock_token
+        auth = FabAuth()
+        with pytest.raises(FabricCLIError, match="missing identity claims"):
+            auth.set_azure_cli()
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_login_rejects_token_missing_tid(self, mock_class, temp_dir_fixture):
+        """set_azure_cli should fail if probe token lacks tid."""
+        header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode()
+        payload = base64.urlsafe_b64encode(
+            _json.dumps({"oid": "o1", "iss": "https://sts.windows.net/t1/"}).encode()
+        ).rstrip(b"=").decode()
+        token_str = f"{header}.{payload}.fakesig"
+        mock_token = MagicMock()
+        mock_token.token = token_str
+        mock_token.expires_on = int(time.time()) + 3600
+        mock_class.return_value.get_token.return_value = mock_token
+        auth = FabAuth()
+        with pytest.raises(FabricCLIError, match="missing identity claims"):
+            auth.set_azure_cli()
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_login_rejects_malformed_token(self, mock_class, temp_dir_fixture):
+        """set_azure_cli should fail if probe token is not a valid JWT."""
+        mock_token = MagicMock()
+        mock_token.token = "not-a-jwt"
+        mock_token.expires_on = int(time.time()) + 3600
+        mock_class.return_value.get_token.return_value = mock_token
+        auth = FabAuth()
+        with pytest.raises(FabricCLIError, match="missing identity claims"):
+            auth.set_azure_cli()
+
+    @patch("fabric_cli.core.fab_auth.AzureCliCredential")
+    def test_acquisition_rejects_token_missing_claims(self, mock_class, temp_dir_fixture):
+        """Token acquisition should fail if returned token lacks identity claims."""
+        # Login with good token
+        _mock_credential_with_jwt(mock_class)
+        auth = FabAuth()
+        auth.set_azure_cli()
+
+        # Now return a bad token on next call
+        bad_token = MagicMock()
+        bad_token.token = "not-a-jwt"
+        bad_token.expires_on = int(time.time()) + 3600
+        mock_class.return_value.get_token.return_value = bad_token
+        with pytest.raises(FabricCLIError, match="missing identity claims"):
+            auth.acquire_token(con.SCOPE_FABRIC_DEFAULT)
+
+
 class TestNonAzureCliIsolation:
     """Verify each auth method uses only its own credential path — no overlap."""