[OPS Common SDK] Update Communication Common SDK for Teams Phone Extensibility GA

sdk/communication/azure-communication-identity/azure/communication/identity/__init__.py

@@ -20,6 +20,8 @@
     PhoneNumberIdentifier,
     PhoneNumberProperties,
     UnknownIdentifier,
+    TeamsExtensionUserIdentifier,
+    TeamsExtensionUserProperties,
 )
 
 __all__ = [
@@ -40,4 +42,6 @@
     "PhoneNumberIdentifier",
     "PhoneNumberProperties",
     "UnknownIdentifier",
+    "TeamsExtensionUserIdentifier",
+    "TeamsExtensionUserProperties",
 ]

sdk/communication/azure-communication-identity/azure/communication/identity/_shared/entra_token_guard_policy.py

@@ -0,0 +1,88 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+import json
+from datetime import datetime, timezone
+from azure.core.pipeline.policies import HTTPPolicy, AsyncHTTPPolicy
+from azure.core.pipeline import PipelineRequest
+
+from dateutil import parser as dateutil_parser
+
+class EntraTokenGuardPolicy(HTTPPolicy):
+    """A pipeline policy that caches the response for a given Entra token and reuses it if valid."""
+
+    def __init__(self):
+        super().__init__()
+        self._entra_token_cache = None
+        self._response_cache = None
+
+    def send(self, request: PipelineRequest):
+        cache_valid, token = _EntraTokenGuardUtils.is_entra_token_cache_valid(self._entra_token_cache, request)
+        if cache_valid and _EntraTokenGuardUtils.is_acs_token_cache_valid(self._response_cache):
+            response = self._response_cache
+        else:
+            self._entra_token_cache = token
+            response = self.next.send(request)
+            self._response_cache = response
+        if response is None:
+            raise RuntimeError("Failed to obtain a valid PipelineResponse in AsyncEntraTokenGuardPolicy.send")
+        return response
+
+
+
+class AsyncEntraTokenGuardPolicy(AsyncHTTPPolicy):
+    """Async pipeline policy that caches the response for a given Entra token and reuses it if valid."""
+
+    def __init__(self):
+        super().__init__()
+        self._entra_token_cache = None
+        self._response_cache = None
+
+    async def send(self, request: PipelineRequest):
+        cache_valid, token = _EntraTokenGuardUtils.is_entra_token_cache_valid(self._entra_token_cache, request)
+        if cache_valid and _EntraTokenGuardUtils.is_acs_token_cache_valid(self._response_cache):
+            response = self._response_cache
+        else:
+            self._entra_token_cache = token
+            response = await self.next.send(request)
+            self._response_cache = response
+        if response is None:
+            raise RuntimeError("Failed to obtain a valid PipelineResponse in AsyncEntraTokenGuardPolicy.send")
+        return response
+
+
+
+class _EntraTokenGuardUtils:
+        @staticmethod
+        def is_entra_token_cache_valid(entra_token_cache, request):
+            current_entra_token = request.http_request.headers.get("Authorization", "")
+            cache_valid = (
+                entra_token_cache is not None and
+                current_entra_token == entra_token_cache
+            )
+            return cache_valid, current_entra_token
+
+        @staticmethod
+        def is_acs_token_cache_valid(response_cache):
+            if response_cache is None or response_cache.http_response.status_code != 200:
+                return False
+            return _EntraTokenGuardUtils.is_access_token_valid(response_cache)
+
+        @staticmethod
+        def is_access_token_valid(response_cache):
+            try:
+                if response_cache is None or response_cache.http_response is None:
+                    return False
+                content = response_cache.http_response.text()
+                data = json.loads(content)
+                expires_on = data["accessToken"]["expiresOn"]
+
+                expires_on_dt = dateutil_parser.parse(expires_on)
+                return datetime.now(timezone.utc) < expires_on_dt
+            except Exception:
+                return False
+
+

sdk/communication/azure-communication-identity/azure/communication/identity/_shared/manual_test.py

@@ -0,0 +1,66 @@
+# coding: utf-8
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+# from azure.core.credentials import AccessToken
+from token_exchange import TokenExchangeClient
+from azure.core.credentials import AccessToken
+from user_credential import CommunicationTokenCredential
+#from user_credential_async import CommunicationTokenCredential
+import asyncio
+
+
+def main():
+    # TODO: Replace these values with your real endpoint and customer token
+    endpoint = "https://acs-auth-ppe-us-ops.unitedstates.ppe.communication.azure.net"
+    customer_token = ""
+
+    # Set up the required options for TokenExchangeClient
+    class CustomTokenCredential:
+        def get_token(self, *scopes, **kwargs):
+            # Provide a hardcoded token and expiry (e.g., far future timestamp)
+            return AccessToken("entra_token", 9999999999)
+
+
+        #token_credential=InteractiveBrowserCredential(redirect_uri="http://localhost", tenant_id="be5c2424-1562-4d62-8d98-815720d06e4a", client_id="c6ec4113-4e29-48a9-a814-e9df50ca033e")
+
+
+    #credentialOld = CommunicationTokenCredential(token="skypetoken")
+    # Example usage of user_credential_async.get_token (assuming user_credential_async is defined/imported)
+    # Initialize the credential with resource_endpoint and custom token_credential
+    credential = CommunicationTokenCredential(
+        resource_endpoint=endpoint,
+        token_credential=CustomTokenCredential(),
+        scopes=["https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls"]
+    )
+
+        # Example call to get_token (update scopes as needed)
+    token = credential.get_token("https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls")
+    print("User Credential Async Token:", token.token)
+
+    token = credential.get_token("https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls")
+    print("User Credential Async Token:", token.token)
+
+    async def get_async_token():
+        token = await credential.get_token("https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls")
+        print("User Credential Async Token (async):", token.token)
+
+        token = await credential.get_token("https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls")
+        print("User Credential Async Token (async):", token.token)
+
+    asyncio.run(get_async_token())
+
+    client = TokenExchangeClient(resource_endpoint=endpoint, token_credential=CustomTokenCredential(), scopes=["https://auth.msft.communication.azure.com/TeamsExtension.ManageCalls"],)
+    try:
+        access_token = client.exchange_entra_token()
+        print("Access Token:", access_token.token)
+        print("Expires On:", access_token.expires_on)
+    except Exception as ex:
+        print("Token exchange failed:", ex)
+
+
+if __name__ == "__main__":
+    main()

sdk/communication/azure-communication-identity/azure/communication/identity/_shared/models.py

@@ -37,6 +37,7 @@ class CommunicationIdentifierKind(str, Enum, metaclass=DeprecatedEnumMeta):
     PHONE_NUMBER = "phone_number"
     MICROSOFT_TEAMS_USER = "microsoft_teams_user"
     MICROSOFT_TEAMS_APP = "microsoft_teams_app"
+    TEAMS_EXTENSION_USER = "teams_extension_user"
 
 
 class CommunicationCloudEnvironment(str, Enum, metaclass=CaseInsensitiveEnumMeta):
@@ -127,6 +128,10 @@ class PhoneNumberProperties(TypedDict):
 
     value: str
     """The phone number in E.164 format."""
+    asserted_id: Optional[str]
+    """The asserted Id set on a phone number to distinguish from other connections made through the same number."""
+    is_anonymous: Optional[bool]
+    """True if the phone number is anonymous, e.g. when used to represent a hidden caller Id."""
 
 
 class PhoneNumberIdentifier:
@@ -139,16 +144,30 @@ class PhoneNumberIdentifier:
     raw_id: str
     """The raw ID of the identifier."""
 
+    PHONE_NUMBER_ANONYMOUS_SUFFIX = "anonymous"
+
     def __init__(self, value: str, **kwargs: Any) -> None:
         """
         :param str value: The phone number.
         :keyword str raw_id: The raw ID of the identifier. If not specified, this will be constructed from
           the 'value' parameter.
         """
-        self.properties = PhoneNumberProperties(value=value)
+
         raw_id: Optional[str] = kwargs.get("raw_id")
+        asserted_id: Optional[str] = None
+        is_anonymous: Optional[bool] = False
+
+        if raw_id is not None:
+            phone_number = raw_id[len(PHONE_NUMBER_PREFIX):]
+            is_anonymous = phone_number == PhoneNumberIdentifier.PHONE_NUMBER_ANONYMOUS_SUFFIX
+            asserted_id_index = -1 if is_anonymous else phone_number.rfind("_") + 1
+            has_asserted_id = 0 < asserted_id_index < len(phone_number)
+            asserted_id = phone_number[asserted_id_index:] if has_asserted_id else None
+
+        self.properties = PhoneNumberProperties(value=value, asserted_id=asserted_id, is_anonymous=is_anonymous)
         self.raw_id = raw_id if raw_id is not None else self._format_raw_id(self.properties)
 
+
     def __eq__(self, other):
         try:
             if other.raw_id:
@@ -163,6 +182,13 @@ def _format_raw_id(self, properties: PhoneNumberProperties) -> str:
         value = properties["value"]
         return f"{PHONE_NUMBER_PREFIX}{value}"
 
+    @property
+    def asserted_id(self) -> Optional[str]:
+        return self.properties.get("asserted_id")
+
+    @property
+    def is_anonymous(self) -> bool:
+        return bool(self.properties.get("is_anonymous"))
 
 class UnknownIdentifier:
     """Represents an identifier of an unknown type.
@@ -347,6 +373,80 @@ def __init__(self, bot_id, **kwargs):
         super().__init__(bot_id, **kwargs)
 
 
+class TeamsExtensionUserProperties(TypedDict):
+    """Dictionary of properties for a TeamsExtensionUserIdentifier."""
+
+    user_id: str
+    """The id of the Teams extension user."""
+    tenant_id: str
+    """The tenant id associated with the user."""
+    resource_id: str
+    """The resource id associated with the user."""
+    cloud: Union[CommunicationCloudEnvironment, str]
+    """Cloud environment that this identifier belongs to."""
+
+
+class TeamsExtensionUserIdentifier:
+    """Represents an identifier for a Teams Extension user."""
+
+    kind: Literal[CommunicationIdentifierKind.TEAMS_EXTENSION_USER] = CommunicationIdentifierKind.TEAMS_EXTENSION_USER
+    """The type of identifier."""
+    properties: TeamsExtensionUserProperties
+    """The properties of the identifier."""
+    raw_id: str
+    """The raw ID of the identifier."""
+
+    def __init__(self, user_id: str, tenant_id: str, resource_id: str, cloud: Union[CommunicationCloudEnvironment, str], **kwargs: Any) -> None:
+        """
+        :param str user_id: Teams extension user id.
+        :param str tenant_id: Tenant id associated with the user.
+        :param str resource_id: Resource id associated with the user.
+        :param cloud: Cloud environment that the user belongs to.
+        :keyword str raw_id: The raw ID of the identifier. If not specified, this value will be constructed from the other properties.
+        """
+        self.properties = TeamsExtensionUserProperties(
+            user_id=user_id,
+            tenant_id=tenant_id,
+            resource_id=resource_id,
+            cloud=cloud,
+        )
+        raw_id: Optional[str] = kwargs.get("raw_id")
+        self.raw_id = raw_id if raw_id is not None else self._format_raw_id(self.properties)
+
+    def __eq__(self, other):
+        try:
+            if other.raw_id:
+                return self.raw_id == other.raw_id
+            return self.raw_id == self._format_raw_id(other.properties)
+        except Exception:  # pylint: disable=broad-except
+            return False
+
+    def _format_raw_id(self, properties: TeamsExtensionUserProperties) -> str:
+        # The prefix depends on the cloud
+        cloud = properties["cloud"]
+        if cloud == CommunicationCloudEnvironment.DOD:
+            prefix = ACS_USER_DOD_CLOUD_PREFIX
+        elif cloud == CommunicationCloudEnvironment.GCCH:
+            prefix = ACS_USER_GCCH_CLOUD_PREFIX
+        else:
+            prefix = ACS_USER_PREFIX
+        return f"{prefix}{properties['resource_id']}_{properties['tenant_id']}_{properties['user_id']}"
+
+def try_create_teams_extension_user(prefix: str, suffix: str) -> Optional[TeamsExtensionUserIdentifier]:
+    segments = suffix.split("_")
+    if len(segments) != 3:
+        return None
+    resource_id, tenant_id, user_id = segments
+    if prefix == ACS_USER_PREFIX:
+        cloud = CommunicationCloudEnvironment.PUBLIC
+    elif prefix == ACS_USER_DOD_CLOUD_PREFIX:
+        cloud = CommunicationCloudEnvironment.DOD
+    elif prefix == ACS_USER_GCCH_CLOUD_PREFIX:
+        cloud = CommunicationCloudEnvironment.GCCH
+    else:
+        raise ValueError(f"Invalid prefix {prefix} for TeamsExtensionUserIdentifier")
+    return TeamsExtensionUserIdentifier(user_id, tenant_id, resource_id, cloud)
+
 def identifier_from_raw_id(raw_id: str) -> CommunicationIdentifier:  # pylint: disable=too-many-return-statements
     """
     Creates a CommunicationIdentifier from a given raw ID.
@@ -407,11 +507,16 @@ def identifier_from_raw_id(raw_id: str) -> CommunicationIdentifier:  # pylint: d
             cloud=CommunicationCloudEnvironment.GCCH,
             raw_id=raw_id,
         )
+    if prefix == SPOOL_USER_PREFIX:
+        return CommunicationUserIdentifier(id=raw_id, raw_id=raw_id)
+
     if prefix in [
         ACS_USER_PREFIX,
         ACS_USER_DOD_CLOUD_PREFIX,
         ACS_USER_GCCH_CLOUD_PREFIX,
-        SPOOL_USER_PREFIX,
     ]:
+        identifier = try_create_teams_extension_user(prefix, suffix)
+        if identifier is not None:
+            return identifier
         return CommunicationUserIdentifier(id=raw_id, raw_id=raw_id)
     return UnknownIdentifier(identifier=raw_id)