twilio · AsabuHere · Dec 12, 2024 · Jul 23, 2024 · Sep 17, 2024 · Sep 17, 2024
diff --git a/twilio/authStrategy/__init__.py b/twilio/authStrategy/__init__.py
diff --git a/twilio/authStrategy/authStrategy.py b/twilio/authStrategy/authStrategy.py
@@ -0,0 +1,22 @@
+from twilio.authStrategy.authType import AuthType
+from enum import Enum
+from abc import abstractmethod
+
+
+class AuthStrategy(object):
+    def __init__(self, auth_type: AuthType):
+        self._auth_type = auth_type
+
+    @property
+    def auth_type(self) -> AuthType:
+        return self._auth_type
+
+    @abstractmethod
+    def get_auth_string(self) -> str:
+        """Return the authentication string."""
+        pass
+
+    @abstractmethod
+    def requires_authentication(self) -> bool:
+        """Return True if authentication is required, else False."""
+        pass
diff --git a/twilio/authStrategy/authType.py b/twilio/authStrategy/authType.py
@@ -0,0 +1,11 @@
+from enum import Enum
+
+class AuthType(Enum):
+    TOKEN = 'token'
+    NO_AUTH = 'noauth'
+    BASIC = 'basic'
+    API_KEY = 'api_key'
+    CLIENT_CREDENTIALS = 'client_credentials'
+
+    def __str__(self):
+        return self.value
diff --git a/twilio/authStrategy/noAuthStrategy.py b/twilio/authStrategy/noAuthStrategy.py
@@ -0,0 +1,11 @@
+from auth_type import AuthType
+
+class NoAuthStrategy(AuthStrategy):
+    def __init__(self):
+        super().__init__(AuthType.NO_AUTH)
+
+    def get_auth_string(self) -> str:
+        return ""
+
+    def requires_authentication(self) -> bool:
+        return False
diff --git a/twilio/authStrategy/tokenAuthStrategy.py b/twilio/authStrategy/tokenAuthStrategy.py
@@ -0,0 +1,39 @@
+import jwt
+import threading
+from datetime import datetime, timedelta
+
+from twilio.authStrategy.authType import AuthType
+from twilio.authStrategy.authStrategy import AuthStrategy
+from twilio.http.token_manager import TokenManager
+
+
+class TokenAuthStrategy(AuthStrategy):
+    def __init__(self, token_manager: TokenManager):
+        super().__init__(AuthType.TOKEN)
+        self.token_manager = token_manager
+        self.token = None
+        self.lock = threading.Lock()
+
+    def get_auth_string(self) -> str:
+        if self.token is None:
+            self.fetch_token()
+        return f"Bearer {self.token}"
+
+    def requires_authentication(self) -> bool:
+        return True
+
+    def fetch_token(self):
+        print(f'token is fetch_token {self.token}')
+        if self.token is None or self.token == "" or self.is_token_expired(self.token):
+            with self.lock:
+                if self.token is None or self.token == "" or self.is_token_expired(self.token):
+                    self.token = self.token_manager.fetch_access_token()
+
+    def is_token_expired(self, token):
+        print(f'token is {token}')
+        decoded_jwt = jwt.decode(token, options={"verify_signature": True}, algorithms=["RS256"])
+        expires_at = decoded_jwt.get("exp")
+        # Add a buffer of 30 seconds
+        buffer_seconds = 30
+        buffer_expires_at = expires_at - buffer_seconds
+        return buffer_expires_at < datetime.datetime.now().timestamp()
diff --git a/twilio/base/client_base.py b/twilio/base/client_base.py
@@ -4,10 +4,11 @@
 from urllib.parse import urlparse, urlunparse
 
 from twilio import __version__
-from twilio.base.exceptions import TwilioException
 from twilio.http import HttpClient
 from twilio.http.http_client import TwilioHttpClient
 from twilio.http.response import Response
+from twilio.authStrategy.authType import AuthType
+from twilio.credential.credentialProvider import CredentialProvider
 
 
 class ClientBase(object):
@@ -23,6 +24,7 @@ def __init__(
         environment: Optional[MutableMapping[str, str]] = None,
         edge: Optional[str] = None,
         user_agent_extensions: Optional[List[str]] = None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initializes the Twilio Client
@@ -35,7 +37,9 @@ def __init__(
         :param environment: Environment to look for auth details, defaults to os.environ
         :param edge: Twilio Edge to make requests to, defaults to None
         :param user_agent_extensions: Additions to the user agent string
+        :param credential_provider: credential provider for authentication method that needs to be used
         """
+
         environment = environment or os.environ
 
         self.username = username or environment.get("TWILIO_ACCOUNT_SID")
@@ -48,9 +52,8 @@ def __init__(
         """ :type : str """
         self.user_agent_extensions = user_agent_extensions or []
         """ :type : list[str] """
-
-        if not self.username or not self.password:
-            raise TwilioException("Credentials are required to create a TwilioClient")
+        self.credential_provider = credential_provider or None
+        """ :type : CredentialProvider """
 
         self.account_sid = account_sid or self.username
         """ :type : str """
@@ -85,8 +88,19 @@ def request(
 
         :returns: Response from the Twilio API
         """
-        auth = self.get_auth(auth)
+
         headers = self.get_headers(method, headers)
+
+        ##If credential provider is provided by user, get the associated auth strategy
+        ##Using the auth strategy, fetch the auth string and set it to authorization header
+        auth_strategy = None ##Initialization
+        if self.credential_provider:
+            auth_strategy = self.credential_provider.to_auth_strategy()
+            headers["Authorization"] = auth_strategy.get_auth_string()
+        else:
+            auth = self.get_auth(auth)
+
+
         uri = self.get_hostname(uri)
 
         return self.http_client.request(
@@ -132,9 +146,19 @@ async def request_async(
                 "http_client must be asynchronous to support async API requests"
             )
 
-        auth = self.get_auth(auth)
+
         headers = self.get_headers(method, headers)
-        uri = self.get_hostname(uri)
+
+        ##If credential provider is provided by user, get the associated auth strategy
+        ##Using the auth strategy, fetch the auth string and set it to authorization header
+        auth_strategy = None ##Initialization
+        if self.credential_provider:
+            auth_strategy = self.credential_provider.to_auth_strategy()
+            if auth_strategy.auth_type == AuthType.TOKEN:
+                auth_strategy.fetch_token()
+                headers["Authorization"] = auth_strategy.get_auth_string()
+        else:
+            auth = self.get_auth(auth)
 
         return await self.http_client.request(
             method,

diff --git a/twilio/base/version.py b/twilio/base/version.py
@@ -461,7 +461,6 @@ def create(
             timeout=timeout,
             allow_redirects=allow_redirects,
         )
-
         return self._parse_create(method, uri, response)
 
     async def create_async(
@@ -488,5 +487,4 @@ async def create_async(
             timeout=timeout,
             allow_redirects=allow_redirects,
         )
-
         return self._parse_create(method, uri, response)
diff --git a/twilio/credential/__init__.py b/twilio/credential/__init__.py
diff --git a/twilio/credential/credentialProvider.py b/twilio/credential/credentialProvider.py
@@ -0,0 +1,12 @@
+from twilio.authStrategy.authType import AuthType
+
+class CredentialProvider:
+    def __init__(self, auth_type: AuthType):
+        self._auth_type = auth_type
+
+    @property
+    def auth_type(self) -> AuthType:
+        return self._auth_type
+
+    def to_auth_strategy(self):
+        raise NotImplementedError("Subclasses must implement this method")
diff --git a/twilio/credential/orgsCredentialProvider.py b/twilio/credential/orgsCredentialProvider.py
@@ -0,0 +1,26 @@
+
+
+from twilio.http.orgs_token_manager import OrgTokenManager
+from twilio.base.exceptions import TwilioException
+from twilio.credential.credentialProvider import CredentialProvider
+from twilio.authStrategy.authType import AuthType
+from twilio.authStrategy.tokenAuthStrategy import TokenAuthStrategy
+
+
+class OrgsCredentialProvider(CredentialProvider):
+    def __init__(self, client_id: str, client_secret: str, token_manager=None):
+        super().__init__(AuthType.CLIENT_CREDENTIALS)
+
+        if client_id is None or client_secret is None:
+            raise TwilioException("Invalid credentials passed")
+
+        self.grant_type = "client_credentials"
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.token_manager = token_manager
+
+    def to_auth_strategy(self):
+        if self.token_manager is None:
+            self.token_manager = OrgTokenManager(self.grant_type, self.client_id, self.client_secret)
+
+        return TokenAuthStrategy(self.token_manager)
diff --git a/twilio/http/http_client.py b/twilio/http/http_client.py
@@ -78,6 +78,11 @@ def request(
         elif timeout <= 0:
             raise ValueError(timeout)
 
+        if headers:
+            if "Requires-Authentication" in headers:
+                headers.pop("Requires-Authentication", None)
+                auth = None
+
         kwargs = {
             "method": method.upper(),
             "url": url,
@@ -91,7 +96,6 @@ def request(
         else:
             kwargs["data"] = data
         self.log_request(kwargs)
-
         self._test_only_last_response = None
         session = self.session or Session()
         request = Request(**kwargs)
@@ -102,7 +106,6 @@ def request(
         settings = session.merge_environment_settings(
             prepped_request.url, self.proxy, None, None, None
         )
-
         response = session.send(
             prepped_request,
             allow_redirects=allow_redirects,

diff --git a/twilio/http/orgs_token_manager.py b/twilio/http/orgs_token_manager.py
@@ -0,0 +1,43 @@
+from twilio.base.version import Version
+from twilio.http.token_manager import TokenManager
+from twilio.rest.preview_iam.v1.token import TokenList
+from twilio.rest import Client
+
+
+class OrgTokenManager(TokenManager):
+    """
+    Orgs Token Manager
+    """
+
+    def __init__(
+        self,
+        grant_type: str,
+        client_id: str,
+        client_secret: str,
+        code: str = None,
+        redirect_uri: str = None,
+        audience: str = None,
+        refreshToken: str = None,
+        scope: str = None,
+    ):
+        self.grant_type = grant_type
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.code = code
+        self.redirect_uri = redirect_uri
+        self.audience = audience
+        self.refreshToken = refreshToken
+        self.scope = scope
+        self.client = Client()
+
+    def fetch_access_token(self):
+        token_instance = self.client.preview_iam.v1.token.create(
+            grant_type=self.grant_type,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            code=self.code,
+            redirect_uri=self.redirect_uri,
+            audience=self.audience,
+            scope=self.scope,
+        )
+        return token_instance.access_token
diff --git a/twilio/http/token_manager.py b/twilio/http/token_manager.py
@@ -0,0 +1,7 @@
+from twilio.base.version import Version
+
+
+class TokenManager:
+
+    def fetch_access_token(self, version: Version):
+        pass
diff --git a/twilio/http/token_manager_initializer.py b/twilio/http/token_manager_initializer.py
@@ -0,0 +1,16 @@
+from twilio.http.token_manager import TokenManager
+
+
+class TokenManagerInitializer:
+
+    org_token_manager = None
+
+    @classmethod
+    def set_token_manager(cls, token_manager: TokenManager):
+        cls.org_token_manager = token_manager
+
+    @classmethod
+    def get_token_manager(cls):
+        if cls.org_token_manager is None:
+            raise Exception("Token Manager not initialized")
+        return cls.org_token_manager
diff --git a/twilio/rest/__init__.py b/twilio/rest/__init__.py
@@ -96,6 +96,7 @@ def __init__(
         environment=None,
         edge=None,
         user_agent_extensions=None,
+        credential_provider=None,
     ):
         """
         Initializes the Twilio Client
@@ -121,6 +122,7 @@ def __init__(
             environment,
             edge,
             user_agent_extensions,
+            credential_provider,
         )
 
         # Domains
@@ -134,7 +136,7 @@ def __init__(
         self._events: Optional["Events"] = None
         self._flex_api: Optional["FlexApi"] = None
         self._frontline_api: Optional["FrontlineApi"] = None
-        self._iam: Optional["Iam"] = None
+        self._preview_iam: Optional["PreviewIam"] = None
         self._insights: Optional["Insights"] = None
         self._intelligence: Optional["Intelligence"] = None
         self._ip_messaging: Optional["IpMessaging"] = None
@@ -147,6 +149,7 @@ def __init__(
         self._numbers: Optional["Numbers"] = None
         self._oauth: Optional["Oauth"] = None
         self._preview: Optional["Preview"] = None
+        self._preview_iam: Optional["PreviewIam"] = None
         self._pricing: Optional["Pricing"] = None
         self._proxy: Optional["Proxy"] = None
         self._routes: Optional["Routes"] = None
@@ -396,6 +399,19 @@ def microvisor(self) -> "Microvisor":
             self._microvisor = Microvisor(self)
         return self._microvisor
 
+    @property
+    def preview_iam(self) -> "PreviewIam":
+        """
+        Access the PreviewIam Twilio Domain
+
+        :returns: PreviewIam Twilio Domain
+        """
+        if self._preview_iam is None:
+            from twilio.rest.preview_iam import PreviewIam
+
+            self._preview_iam = PreviewIam(self)
+        return self._preview_iam
+
     @property
     def monitor(self) -> "Monitor":
         """