Release 2.1.1: backport Entra caller app validation

packages/apps/src/microsoft_teams/apps/auth/token_validator.py

@@ -16,6 +16,8 @@
 JWT_LEEWAY_SECONDS = 300  # Allowable clock skew when validating JWTs
 _MAX_ENTRA_VALIDATOR_CACHE_SIZE = 100
 ENTRA_V1_ISSUER_PREFIX = "https://sts.windows.net/"
+AGENT_365_PLATFORM_APP_ID = "5a807f24-c9de-44ee-a3a7-329e88a00ffc"
+"""App ID of the Agent 365 platform."""
 
 logger = logging.getLogger(__name__)
 
@@ -265,7 +267,20 @@ async def _validate_entra_token(self, raw_token: str, unverified_payload: Dict[s
         validator = self._get_entra_validator(tenant_id)
         # TODO: Agent ID inbound Entra tokens currently do not include serviceurl. Revisit service URL
         # validation for this path once the platform defines a signed service URL claim or equivalent.
-        return await validator.validate_token(raw_token)
+        payload = await validator.validate_token(raw_token)
+        self._validate_entra_caller_app(payload)
+        return payload
+
+    def _validate_entra_caller_app(self, payload: Dict[str, Any]) -> None:
+        """Require the client app that requested the token to be the Agent 365 platform.
+
+        Entra v2 tokens carry the caller in ``azp`` and v1 tokens carry it in ``appid``.
+        ``appid`` is only consulted when ``azp`` is absent, so a present but invalid ``azp`` is rejected.
+        """
+        caller_app_id = payload["azp"] if "azp" in payload else payload.get("appid")
+        if not isinstance(caller_app_id, str) or caller_app_id.lower() != AGENT_365_PLATFORM_APP_ID:
+            logger.warning("Entra inbound token caller app is not allowed: %s", caller_app_id)
+            raise jwt.InvalidTokenError("Entra inbound token caller app is not allowed")
 
     def _get_entra_validator(self, tenant_id: str) -> TokenValidator:
         cached_validator = self._entra_validators_by_tenant.get(tenant_id)

packages/apps/tests/test_token_validator.py

@@ -3,12 +3,17 @@
 Licensed under the MIT License.
 """
 
+from typing import Any, Dict
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import jwt
 import pytest
 from microsoft_teams.apps.auth import token_validator as token_validator_module
-from microsoft_teams.apps.auth.token_validator import InboundActivityTokenValidator, TokenValidator
+from microsoft_teams.apps.auth.token_validator import (
+    AGENT_365_PLATFORM_APP_ID,
+    InboundActivityTokenValidator,
+    TokenValidator,
+)
 
 # pyright: basic
 
@@ -428,7 +433,7 @@ async def test_validate_token_uses_entra_validator_for_v2_issuer(self):
         validator = InboundActivityTokenValidator("test-app-id")
         validator._service_validator.validate_token = AsyncMock()
         entra_validator = MagicMock()
-        entra_validator.validate_token = AsyncMock(return_value={"tid": "tenant-id"})
+        entra_validator.validate_token = AsyncMock(return_value={"tid": "tenant-id", "azp": AGENT_365_PLATFORM_APP_ID})
 
         with patch.object(validator, "_get_entra_validator", return_value=entra_validator) as get_validator:
             with patch(
@@ -437,7 +442,7 @@ async def test_validate_token_uses_entra_validator_for_v2_issuer(self):
             ):
                 result = await validator.validate_token("entra-token", "https://service.example")
 
-        assert result == {"tid": "tenant-id"}
+        assert result == {"tid": "tenant-id", "azp": AGENT_365_PLATFORM_APP_ID}
         get_validator.assert_called_once_with("tenant-id")
         entra_validator.validate_token.assert_called_once_with("entra-token")
         validator._service_validator.validate_token.assert_not_called()
@@ -446,13 +451,15 @@ async def test_validate_token_uses_entra_validator_for_v2_issuer(self):
     async def test_validate_token_uses_entra_validator_for_v1_sts_issuer(self):
         validator = InboundActivityTokenValidator("test-app-id")
         entra_validator = MagicMock()
-        entra_validator.validate_token = AsyncMock(return_value={"tid": "tenant-id"})
+        entra_validator.validate_token = AsyncMock(
+            return_value={"tid": "tenant-id", "appid": AGENT_365_PLATFORM_APP_ID}
+        )
 
         with patch.object(validator, "_get_entra_validator", return_value=entra_validator) as get_validator:
             with patch("jwt.decode", return_value={"iss": "https://sts.windows.net/tenant-id/", "tid": "tenant-id"}):
                 result = await validator.validate_token("entra-v1-token")
 
-        assert result == {"tid": "tenant-id"}
+        assert result == {"tid": "tenant-id", "appid": AGENT_365_PLATFORM_APP_ID}
         get_validator.assert_called_once_with("tenant-id")
         entra_validator.validate_token.assert_called_once_with("entra-v1-token")
 
@@ -499,3 +506,71 @@ def test_get_entra_validator_cache_is_bounded(self):
         assert "tenant-0" not in validator._entra_validators_by_tenant
         last_tenant_id = f"tenant-{token_validator_module._MAX_ENTRA_VALIDATOR_CACHE_SIZE}"
         assert last_tenant_id in validator._entra_validators_by_tenant
+
+
+class TestInboundActivityTokenValidatorEntraCallerApp:
+    OTHER_APP_ID = "00000000-0000-0000-0000-000000000099"
+    ENTRA_CLAIMS = {"iss": "https://login.microsoftonline.com/tenant-id/v2.0", "tid": "tenant-id"}
+
+    async def _validate_entra(self, validated_payload: Dict[str, Any]) -> Dict[str, Any]:
+        validator = InboundActivityTokenValidator("test-app-id")
+        entra_validator = MagicMock()
+        entra_validator.validate_token = AsyncMock(return_value=validated_payload)
+
+        with patch.object(validator, "_get_entra_validator", return_value=entra_validator):
+            with patch("jwt.decode", return_value=self.ENTRA_CLAIMS):
+                return await validator.validate_token("entra-token")
+
+    @pytest.mark.asyncio
+    async def test_accepts_agent_365_platform_app_in_azp(self):
+        payload = {"tid": "tenant-id", "azp": AGENT_365_PLATFORM_APP_ID}
+
+        assert await self._validate_entra(payload) == payload
+
+    @pytest.mark.asyncio
+    async def test_accepts_agent_365_platform_app_in_appid_when_azp_absent(self):
+        payload = {"tid": "tenant-id", "appid": AGENT_365_PLATFORM_APP_ID.upper()}
+
+        assert await self._validate_entra(payload) == payload
+
+    @pytest.mark.asyncio
+    async def test_rejects_other_app_in_azp(self):
+        with pytest.raises(jwt.InvalidTokenError, match="caller app is not allowed"):
+            await self._validate_entra({"tid": "tenant-id", "azp": self.OTHER_APP_ID})
+
+    @pytest.mark.asyncio
+    async def test_rejects_other_app_in_appid(self):
+        with pytest.raises(jwt.InvalidTokenError, match="caller app is not allowed"):
+            await self._validate_entra({"tid": "tenant-id", "appid": self.OTHER_APP_ID})
+
+    @pytest.mark.asyncio
+    async def test_rejects_token_without_azp_or_appid(self):
+        with pytest.raises(jwt.InvalidTokenError, match="caller app is not allowed"):
+            await self._validate_entra({"tid": "tenant-id"})
+
+    @pytest.mark.asyncio
+    async def test_prefers_azp_over_appid(self):
+        with pytest.raises(jwt.InvalidTokenError, match="caller app is not allowed"):
+            await self._validate_entra(
+                {"tid": "tenant-id", "azp": self.OTHER_APP_ID, "appid": AGENT_365_PLATFORM_APP_ID}
+            )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("azp", ["", None, 123])
+    async def test_rejects_present_but_invalid_azp_without_falling_back_to_appid(self, azp: Any):
+        with pytest.raises(jwt.InvalidTokenError, match="caller app is not allowed"):
+            await self._validate_entra({"tid": "tenant-id", "azp": azp, "appid": AGENT_365_PLATFORM_APP_ID})
+
+    @pytest.mark.asyncio
+    async def test_does_not_apply_caller_app_check_to_bot_framework_tokens(self):
+        validator = InboundActivityTokenValidator("test-app-id")
+        bot_payload = {"iss": "https://api.botframework.com", "appid": self.OTHER_APP_ID}
+        validator._service_validator.validate_token = AsyncMock(return_value=bot_payload)
+
+        with patch("jwt.decode", return_value={"iss": "https://api.botframework.com"}):
+            result = await validator.validate_token("bot-token", "https://service.example")
+
+        assert result == bot_payload
+
+    def test_agent_365_platform_app_id_has_expected_value(self):
+        assert AGENT_365_PLATFORM_APP_ID == "5a807f24-c9de-44ee-a3a7-329e88a00ffc"

version.json

@@ -1,6 +1,6 @@
 {
   "$schema": "https://raw.githubusercontent.com/dotnet/Nerdbank.GitVersioning/main/src/NerdBank.GitVersioning/version.schema.json",
-  "version": "2.1.0",
+  "version": "2.1.1",
   "versionHeightOffset": 1,
   "publicReleaseRefSpec": [
     "^refs/heads/release/v\\d+\\.\\d+$"