feat: oauth sdk implementation

twilio/http/bearer_token_http_client.py

@@ -0,0 +1,28 @@
+import datetime
+import jwt
+
+from twilio.base.version import Version
+from twilio.http.orgs_token_manager import OrgTokenManager
+from twilio.twilio_bearer_token_auth import TwilioBearerTokenAuth
+
+
+class BearerTokenHTTPClient:
+    def __init__(self, orgs_token_manager: OrgTokenManager):
+        self.orgs_token_manager = orgs_token_manager
+
+    def get_headers(self, version: Version):
+        if TwilioBearerTokenAuth.get_access_token() is None or self.is_token_expired(TwilioBearerTokenAuth.get_access_token()):
+            access_token = self.orgs_token_manager.fetch_access_token(version)
+            TwilioBearerTokenAuth.init(access_token)
+        else:
+            access_token = TwilioBearerTokenAuth.get_access_token()
+
+        return 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/http/no_auth_http_client.py

@@ -0,0 +1,5 @@
+class NoAuthHTTPClient:
+
+    def get_headers(self):
+        headers = {}
+        return headers

twilio/http/orgs_token_manager.py

@@ -0,0 +1,31 @@
+
+from twilio.base.version import Version
+from twilio.rest.preview_iam.organizations.token import TokenList
+
+class OrgTokenManager():
+    """
+    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
+
+    def fetch_access_token(self, version: Version):
+        token_list = TokenList(version)
+        token_instance = token_list.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

twilio/http/token_manager.py

@@ -0,0 +1,18 @@
+from twilio.base.version import Version
+from twilio.http.orgs_token_manager import OrgTokenManager
+class TokenManager:
+    org_token_manager = None
+
+    @classmethod
+    def init(cls, 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, tokenManager: OrgTokenManager = None):
+        if tokenManager is None:
+            cls.org_token_manager = OrgTokenManager(grant_type, client_id, client_secret, code, redirect_uri, audience, refreshToken, scope)
+        else:
+            cls.org_token_manager = tokenManager
+
+    @classmethod
+    def get_token_manager(cls, version: Version):
+        if cls.org_token_manager is None:
+            raise Exception("Token Manager not initialized")
+        cls.org_token_manager.version = version
+        return cls.org_token_manager

twilio/rest/__init__.py

@@ -145,6 +145,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
@@ -446,6 +447,19 @@ def preview(self) -> "Preview":
             self._preview = Preview(self)
         return self._preview
 
+    @property
+    def preview_iam(self) -> "PreviewIam":
+        """
+        Access the Preview Twilio Domain
+
+        :returns: Preview Twilio Domain
+        """
+        if self._preview_iam is None:
+            from twilio.rest.preview_iam import PreviewIam
+
+            self._preview = PreviewIam(self)
+        return self._preview_iam
+
     @property
     def pricing(self) -> "Pricing":
         """

twilio/rest/preview_iam/PreviewIamBase.py

@@ -0,0 +1,29 @@
+r"""
+  This code was generated by
+  ___ _ _ _ _ _    _ ____    ____ ____ _    ____ ____ _  _ ____ ____ ____ ___ __   __
+   |  | | | | |    | |  | __ |  | |__| | __ | __ |___ |\ | |___ |__/ |__|  | |  | |__/
+   |  |_|_| | |___ | |__|    |__| |  | |    |__] |___ | \| |___ |  \ |  |  | |__| |  \
+
+  NOTE: This class is auto generated by OpenAPI Generator.
+  https://openapi-generator.tech
+  Do not edit the class manually.
+"""
+from twilio.base.domain import Domain
+from twilio.rest import Client
+from twilio.rest.preview_iam.organizations_openapi.token import Token
+from twilio.rest.preview_iam.organizations_openapi.account import Account
+from twilio.rest.preview_iam.organizations_openapi.authorize import AuthorizeList
+from twilio.rest.preview_iam.organizations_openapi.resource_type import ResourceTypeList
+from twilio.rest.preview_iam.organizations_openapi.role_assignment import RoleAssignmentList
+
+class PreviewIamBase(Domain):
+    def __init__(self, twilio: Client):
+        """
+        Initialize the PreviewIam Domain
+
+        :returns: Domain for PreviewIam
+        """
+        super().__init__(twilio, "https://preview.twilio.com/iam")
+        self._token: Optional[TokenList] = None
+        self._service_accounts: Optional[ServiceAccounts] = None
+        self._service_roles: Optional[ServiceRoles] = None

twilio/rest/preview_iam/organizations/token.py

@@ -0,0 +1,160 @@
+r"""
+    This code was generated by
+   ___ _ _ _ _ _    _ ____    ____ ____ _    ____ ____ _  _ ____ ____ ____ ___ __   __
+    |  | | | | |    | |  | __ |  | |__| | __ | __ |___ |\ | |___ |__/ |__|  | |  | |__/
+    |  |_|_| | |___ | |__|    |__| |  | |    |__] |___ | \| |___ |  \ |  |  | |__| |  \
+
+    Organization Public API
+    No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
+
+    NOTE: This class is auto generated by OpenAPI Generator.
+    https://openapi-generator.tech
+    Do not edit the class manually.
+"""
+
+
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator
+from twilio.base import deserialize, serialize, values
+
+from twilio.base.instance_resource import InstanceResource
+from twilio.base.list_resource import ListResource
+from twilio.base.version import Version
+
+
+
+class TokenInstance(InstanceResource):
+
+    """
+    :ivar access_token: Token which carries the necessary information to access a Twilio resource directly.
+    :ivar refresh_token: Token which carries the information necessary to get a new access token.
+    :ivar id_token: Token which carries the information necessary of user profile.
+    :ivar token_type: Token type
+    :ivar expires_in: 
+    """
+
+    def __init__(self, version: Version, payload: Dict[str, Any]):
+        super().__init__(version)
+
+
+        self.access_token: Optional[str] = payload.get("access_token")
+        self.refresh_token: Optional[str] = payload.get("refresh_token")
+        self.id_token: Optional[str] = payload.get("id_token")
+        self.token_type: Optional[str] = payload.get("token_type")
+        self.expires_in: Optional[int] = payload.get("expires_in")
+
+
+
+
+    def __repr__(self) -> str:
+        """
+        Provide a friendly representation
+
+        :returns: Machine friendly representation
+        """
+
+        return '<Twilio.PreviewIam.Organizations.TokenInstance>'
+
+
+
+
+class TokenList(ListResource):
+
+    def __init__(self, version: Version):
+        """
+        Initialize the TokenList
+
+        :param version: Version that contains the resource
+
+        """
+        super().__init__(version)
+
+
+        self._uri = '/token'
+
+
+
+    def create(self, grant_type: str, client_id: str, client_secret: Union[str, object]=values.unset, code: Union[str, object]=values.unset, redirect_uri: Union[str, object]=values.unset, audience: Union[str, object]=values.unset, refresh_token: Union[str, object]=values.unset, scope: Union[str, object]=values.unset) -> TokenInstance:
+        """
+        Create the TokenInstance
+
+        :param grant_type: Grant type is a credential representing resource owner's authorization which can be used by client to obtain access token.
+        :param client_id: A 34 character string that uniquely identifies this OAuth App.
+        :param client_secret: The credential for confidential OAuth App.
+        :param code: JWT token related to the authorization code grant type.
+        :param redirect_uri: The redirect uri
+        :param audience: The targeted audience uri
+        :param refresh_token: JWT token related to refresh access token.
+        :param scope: The scope of token
+
+        :returns: The created TokenInstance
+        """
+
+        data = values.of({ 
+            'grant_type': grant_type,
+            'client_id': client_id,
+            'client_secret': client_secret,
+            'code': code,
+            'redirect_uri': redirect_uri,
+            'audience': audience,
+            'refresh_token': refresh_token,
+            'scope': scope,
+        })
+        headers = values.of({
+                'Content-Type': 'application/x-www-form-urlencoded'
+            })
+
+
+
+        payload = self._version.create(method='POST', uri=self._uri, data=data, headers=headers)
+
+        return TokenInstance(self._version, payload)
+
+    async def create_async(self, grant_type: str, client_id: str, client_secret: Union[str, object]=values.unset, code: Union[str, object]=values.unset, redirect_uri: Union[str, object]=values.unset, audience: Union[str, object]=values.unset, refresh_token: Union[str, object]=values.unset, scope: Union[str, object]=values.unset) -> TokenInstance:
+        """
+        Asynchronously create the TokenInstance
+
+        :param grant_type: Grant type is a credential representing resource owner's authorization which can be used by client to obtain access token.
+        :param client_id: A 34 character string that uniquely identifies this OAuth App.
+        :param client_secret: The credential for confidential OAuth App.
+        :param code: JWT token related to the authorization code grant type.
+        :param redirect_uri: The redirect uri
+        :param audience: The targeted audience uri
+        :param refresh_token: JWT token related to refresh access token.
+        :param scope: The scope of token
+
+        :returns: The created TokenInstance
+        """
+
+        data = values.of({ 
+            'grant_type': grant_type,
+            'client_id': client_id,
+            'client_secret': client_secret,
+            'code': code,
+            'redirect_uri': redirect_uri,
+            'audience': audience,
+            'refresh_token': refresh_token,
+            'scope': scope,
+        })
+        headers = values.of({
+                'Content-Type': 'application/x-www-form-urlencoded'
+            })
+
+
+
+        payload = await self._version.create_async(method='POST', uri=self._uri, data=data, headers=headers)
+
+        return TokenInstance(self._version, payload)
+
+
+
+
+    def __repr__(self) -> str:
+        """
+        Provide a friendly representation
+
+        :returns: Machine friendly representation
+        """
+        return '<Twilio.PreviewIam.Organizations.TokenList>'
+

twilio/twilio_bearer_token_auth.py

@@ -0,0 +1,34 @@
+from threading import Lock
+from concurrent.futures import ThreadPoolExecutor
+
+
+class BearerTokenTwilioRestClient:
+    pass
+
+
+class TwilioBearerTokenAuth:
+    _lock = Lock()
+    access_token = None
+    rest_client = None
+    user_agent_extensions = None
+    region = None
+    edge = None
+
+    @classmethod
+    def init(cls, access_token):
+        with cls._lock:
+            if not access_token:
+                raise ValueError("Access Token cannot be null or Empty")
+            if access_token != cls.access_token:
+                cls.access_token = None
+            cls.access_token = access_token
+
+    @classmethod
+    def get_access_token(cls):
+        with cls._lock:
+            return cls.access_token
+
+    @classmethod
+    def get_header_param(cls):
+        with cls._lock:
+            return {"Authorization": "Bearer {token}".format(token=cls.access_token)}

twilio/twilio_no_auth.py

@@ -0,0 +1,33 @@
+import os
+from threading import Lock
+
+
+class NoAuthTwilioRestClient:
+    pass
+
+
+class TwilioNoAuth:
+    _lock = Lock()
+    no_auth_twilio_rest_client = None
+    user_agent_extensions = None
+    region = os.getenv("TWILIO_REGION")
+    edge = os.getenv("TWILIO_EDGE")
+
+    @classmethod
+    def get_rest_client(cls):
+        with cls._lock:
+            if cls.no_auth_twilio_rest_client is None:
+                cls.no_auth_twilio_rest_client = cls.build_oauth_rest_client()
+        return cls.no_auth_twilio_rest_client
+
+    @classmethod
+    def build_oauth_rest_client(cls):
+        builder = NoAuthTwilioRestClient.Builder()
+
+        if cls.user_agent_extensions is not None:
+            builder.user_agent_extensions(cls.user_agent_extensions)
+
+        builder.region(cls.region)
+        builder.edge(cls.edge)
+
+        return builder.build()