twilio python changes for orgs api uptake

Author: AsabuHere

twilio/authStrategy/authStrategy.py

@@ -0,0 +1,21 @@
+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
+
+    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

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

twilio/authStrategy/basicAuthStrategy.py

@@ -0,0 +1,17 @@
+import base64
+from enum import Enum
+
+
+class BasicAuthStrategy(AuthStrategy):
+    def __init__(self, username: str, password: str):
+        super().__init__(AuthType.BASIC)
+        self.username = username
+        self.password = password
+
+    def get_auth_string(self) -> str:
+        credentials = f"{self.username}:{self.password}"
+        encoded = base64.b64encode(credentials.encode('ascii')).decode('ascii')
+        return f"Basic {encoded}"
+
+    def requires_authentication(self) -> bool:
+        return True

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

twilio/authStrategy/tokenAuthStrategy.py

@@ -0,0 +1,35 @@
+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:
+        return f"Bearer {self.token}"
+
+    def requires_authentication(self) -> bool:
+        return True
+
+    def fetch_token(self):
+        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):
+        decoded_jwt = jwt.decode(token, options={"verify_signature": True})
+        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()

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 """
@@ -69,8 +72,6 @@ def request(
         auth: Optional[Tuple[str, str]] = None,
         timeout: Optional[float] = None,
         allow_redirects: bool = False,
-        is_oauth: bool = False,
-        domain: Optional[str] = None,
     ) -> Response:
         """
         Makes a request to the Twilio API using the configured http client
@@ -88,20 +89,26 @@ def request(
         :returns: Response from the Twilio API
         """
 
-        print('*****')
-        if not is_oauth:
-            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:
+            print(f'Reached here 2 {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()
+            if auth_strategy.auth_type == AuthType.BASIC:
+                headers["Authorization"] = auth_strategy.get_auth_string()
+        else:
+            auth = self.get_auth(auth)
+
+            print(f'auth2 *** {auth}')
+
+
         uri = self.get_hostname(uri)
-        if is_oauth:
-            OauthTokenBase = dynamic_import(
-                "twilio.base.oauth_token_base", "OauthTokenBase"
-            )
-            token = OauthTokenBase().get_oauth_token(
-                domain, "v1", self.username, self.password
-            )
-            headers["Authorization"] = f"Bearer {token}"
-            headers.get("Authorization")
 
         return self.http_client.request(
             method,
@@ -124,7 +131,6 @@ async def request_async(
         auth: Optional[Tuple[str, str]] = None,
         timeout: Optional[float] = None,
         allow_redirects: bool = False,
-        is_oauth: bool = False,
     ) -> Response:
         """
         Asynchronously makes a request to the Twilio API  using the configured http client
@@ -146,19 +152,25 @@ async def request_async(
             raise RuntimeError(
                 "http_client must be asynchronous to support async API requests"
             )
-        if not is_oauth:
-            auth = self.get_auth(auth)
+
+
         headers = self.get_headers(method, headers)
-        uri = self.get_hostname(uri)
-        if is_oauth:
-            OauthTokenBase = dynamic_import(
-                "twilio.base.oauth_token_base", "OauthTokenBase"
-            )
-            token = OauthTokenBase().get_oauth_token(
-                domain, "v1", self.username, self.password
-            )
-            headers["Authorization"] = f"Bearer {token}"
-            headers.get("Authorization")
+
+        ##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:
+            print(f'Reached here 1')
+            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()
+            if auth_strategy.auth_type == AuthType.BASIC:
+                headers["Authorization"] = auth_strategy.get_auth_string()
+        else:
+            auth = self.get_auth(auth)
+
+        print(f'auth2 *** {auth}')
 
         return await self.http_client.request(
             method,

twilio/base/domain.py

@@ -32,7 +32,6 @@ def request(
         auth: Optional[Tuple[str, str]] = None,
         timeout: Optional[float] = None,
         allow_redirects: bool = False,
-        is_oauth: bool = False,
     ) -> Response:
         """
         Makes an HTTP request to this domain.
@@ -56,8 +55,6 @@ def request(
             auth=auth,
             timeout=timeout,
             allow_redirects=allow_redirects,
-            is_oauth=is_oauth,
-            domain=self.base_url,
         )
 
     async def request_async(
@@ -70,7 +67,6 @@ async def request_async(
         auth: Optional[Tuple[str, str]] = None,
         timeout: Optional[float] = None,
         allow_redirects: bool = False,
-        is_oauth: bool = False,
     ) -> Response:
         """
         Makes an asynchronous HTTP request to this domain.
@@ -94,5 +90,4 @@ async def request_async(
             auth=auth,
             timeout=timeout,
             allow_redirects=allow_redirects,
-            is_oauth=is_oauth,
         )

twilio/base/exceptions.py

@@ -63,7 +63,6 @@ def get_uri(code: int) -> str:
                     red_error=red("HTTP Error"),
                     request_was=white("Your request was:"),
                     http_line=teal("%s %s" % (self.method, self.uri, self.data, self.uri)),
-                    http_line=teal("%s %s" % (self.data, self.headers)),
                     twilio_returned=white("Twilio returned the following information:"),
                     message=blue(str(self.msg)),
                 )