Add auth files

Author: heyitsaamir

packages/api/pyproject.toml

@@ -13,6 +13,7 @@ requires-python = ">=3.12"
 dependencies = [
     "pydantic>=2.0.0",
     "microsoft-teams-common",
+    "jwt>=1.3.1",
 ]
 
 [tool.uv.sources]

packages/api/src/microsoft/teams/api/__init__.py

@@ -3,13 +3,16 @@
 Licensed under the MIT License.
 """
 
+from .auth import *  # noqa: F403
+from .auth import __all__ as auth_all
 from .clients import *  # noqa: F403
 from .clients import __all__ as clients_all
 from .models import *  # noqa: F403
 from .models import __all__ as models_all
 
 # Combine all exports from submodules
 __all__ = [
+    *auth_all,
     *clients_all,
     *models_all,
 ]

packages/api/src/microsoft/teams/api/auth/__init__.py

@@ -0,0 +1,20 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from .caller import CALLER_IDS, CallerType
+from .credentials import ClientCredentials, Credentials, TokenCredentials
+from .json_web_token import JsonWebToken, JsonWebTokenPayload
+from .token import IToken
+
+__all__ = [
+    "CALLER_IDS",
+    "CallerType",
+    "ClientCredentials",
+    "Credentials",
+    "TokenCredentials",
+    "IToken",
+    "JsonWebToken",
+    "JsonWebTokenPayload",
+]

packages/api/src/microsoft/teams/api/auth/caller.py

@@ -0,0 +1,14 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from typing import Literal
+
+CallerType = Literal["azure", "gov", "bot"]
+
+CALLER_IDS = {
+    "azure": "urn:botframework:azure",
+    "gov": "urn:botframework:azureusgov",
+    "bot": "urn:botframework:aadappid",
+}

packages/api/src/microsoft/teams/api/models/auth/credentials.py renamed to packages/api/src/microsoft/teams/api/auth/credentials.py

@@ -0,0 +1,129 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import time
+from typing import Optional, Union
+
+import jwt
+from pydantic import BaseModel, ConfigDict
+
+from .caller import CALLER_IDS, CallerType
+from .token import IToken
+
+
+class JsonWebTokenPayload(BaseModel):
+    """JWT payload with additional Teams-specific fields."""
+
+    model_config = ConfigDict(extra="allow")
+
+    aud: Optional[Union[str, list[str]]] = None
+    iss: Optional[str] = None
+    exp: Optional[int] = None
+    kid: Optional[str] = None
+    appid: Optional[str] = None
+    app_displayname: Optional[str] = None
+    tid: Optional[str] = None
+    version: Optional[str] = None
+    serviceurl: Optional[str] = None
+
+
+class JsonWebToken(IToken):
+    """JSON Web Token implementation for Teams authentication."""
+
+    def __init__(self, value: str):
+        """
+        Initialize JWT from token string.
+
+        Args:
+            value: The JWT token string.
+        """
+        self._value = value
+        # Decode without verification for payload extraction
+        jwt_instance = jwt.JWT()
+        self._payload = JsonWebTokenPayload(**jwt_instance.decode(value, do_verify=False, do_time_check=False))
+
+    @property
+    def audience(self) -> Optional[Union[str, list[str]]]:
+        """The token audience."""
+        return self._payload.aud
+
+    @property
+    def issuer(self) -> Optional[str]:
+        """The token issuer."""
+        return self._payload.iss
+
+    @property
+    def key_id(self) -> Optional[str]:
+        """The key ID."""
+        return self._payload.kid
+
+    @property
+    def app_id(self) -> str:
+        """The app ID."""
+        return self._payload.appid or ""
+
+    @property
+    def app_display_name(self) -> Optional[str]:
+        """The app display name."""
+        return self._payload.app_displayname
+
+    @property
+    def tenant_id(self) -> Optional[str]:
+        """The tenant ID."""
+        return self._payload.tid
+
+    @property
+    def version(self) -> Optional[str]:
+        """The token version."""
+        return self._payload.version
+
+    @property
+    def service_url(self) -> str:
+        """The service URL to send responses to."""
+        url = self._payload.serviceurl or "https://smba.trafficmanager.net/teams"
+
+        if url.endswith("/"):
+            url = url[:-1]
+
+        return url
+
+    @property
+    def from_(self) -> CallerType:
+        """Where the activity originated from."""
+        if self.app_id:
+            return "bot"
+        return "azure"
+
+    @property
+    def from_id(self) -> str:
+        """The id of the activity sender."""
+        if self.from_ == "bot":
+            return f"{CALLER_IDS['bot']}:{self.app_id}"
+        return CALLER_IDS["azure"]
+
+    @property
+    def expiration(self) -> Optional[int]:
+        """The expiration of the token since epoch in milliseconds."""
+        if self._payload.exp:
+            return self._payload.exp * 1000
+        return None
+
+    def is_expired(self, buffer_ms: int = 5 * 60 * 1000) -> bool:
+        """
+        Check if the token is expired.
+
+        Args:
+            buffer_ms: Buffer time in milliseconds (default 5 minutes).
+
+        Returns:
+            True if the token is expired, False otherwise.
+        """
+        if not self.expiration:
+            return False
+        return self.expiration < (time.time() * 1000) + buffer_ms
+
+    def __str__(self) -> str:
+        """String form of the token."""
+        return self._value

packages/api/src/microsoft/teams/api/auth/json_web_token.py

@@ -0,0 +1,73 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from .caller import CallerType
+
+
+class IToken(ABC):
+    """Any authorized token."""
+
+    @property
+    @abstractmethod
+    def app_id(self) -> str:
+        """The app id."""
+        pass
+
+    @property
+    @abstractmethod
+    def app_display_name(self) -> Optional[str]:
+        """The app display name."""
+        pass
+
+    @property
+    @abstractmethod
+    def tenant_id(self) -> Optional[str]:
+        """The tenant id."""
+        pass
+
+    @property
+    @abstractmethod
+    def service_url(self) -> str:
+        """The service url to send responses to."""
+        pass
+
+    @property
+    @abstractmethod
+    def from_(self) -> CallerType:
+        """Where the activity originated from."""
+        pass
+
+    @property
+    @abstractmethod
+    def from_id(self) -> str:
+        """The id of the activity sender."""
+        pass
+
+    @property
+    @abstractmethod
+    def expiration(self) -> Optional[int]:
+        """The expiration of the token since epoch in milliseconds."""
+        pass
+
+    @abstractmethod
+    def is_expired(self, buffer_ms: int = 5 * 60 * 1000) -> bool:
+        """
+        Check if the token is expired.
+
+        Args:
+            buffer_ms: Buffer time in milliseconds (default 5 minutes).
+
+        Returns:
+            True if the token is expired, False otherwise.
+        """
+        pass
+
+    @abstractmethod
+    def __str__(self) -> str:
+        """String form of the token."""
+        pass

packages/api/src/microsoft/teams/api/auth/token.py

@@ -10,7 +10,7 @@
 from pydantic import AliasGenerator, BaseModel, ConfigDict
 from pydantic.alias_generators import to_camel
 
-from ...models import Credentials, TokenCredentials
+from ...auth import Credentials, TokenCredentials
 from ..base_client import BaseClient
 
 

packages/api/src/microsoft/teams/api/clients/bot/token_client.py

@@ -5,8 +5,6 @@
 
 from .account import Account, AccountRole
 from .activity import Activity
-from .auth import *  # noqa: F403
-from .auth import __all__ as auth_all
 from .channel_id import ChannelID
 from .conversation import *  # noqa: F403
 from .conversation import __all__ as conversation_all
@@ -25,6 +23,5 @@
     *conversation_all,
     *sign_in_all,
     *token_all,
-    *auth_all,
     *token_exchange_all,
 ]