From 03260ac88913af400a493a43350b920642f77c71 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 21:57:47 +0200 Subject: [PATCH 01/21] Add auth utils --- seam/utils/auth.py | 246 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 seam/utils/auth.py diff --git a/seam/utils/auth.py b/seam/utils/auth.py new file mode 100644 index 00000000..fa6337cb --- /dev/null +++ b/seam/utils/auth.py @@ -0,0 +1,246 @@ +from typing import Optional +import re +from seam.utils.options import ( + SeamHttpInvalidOptionsError, + is_seam_http_multi_workspace_options_with_personal_access_token, + is_seam_http_options_with_api_key, + is_seam_http_options_with_client_session_token, + is_seam_http_multi_workspace_options_with_console_session_token, + is_seam_http_options_with_console_session_token, + is_seam_http_options_with_personal_access_token, +) +from seam.utils.token import ( + is_jwt, + is_access_token, + is_client_session_token, + is_publishable_key, + is_seam_token, + publishable_key_token_prefix, + token_prefix, + client_session_token_prefix, + jwt_prefix, + access_token_prefix, +) + + +class SeamHttpInvalidTokenError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received an invalid token: {message}") + + +def get_auth_headers( + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, +): + if publishable_key: + return get_auth_headers_for_publishable_key(publishable_key) + + if is_seam_http_options_with_api_key( + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ): + return get_auth_headers_for_api_key(api_key) + + if is_seam_http_options_with_client_session_token( + client_session_token=client_session_token, + api_key=api_key, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ): + return get_auth_headers_for_client_session_token(client_session_token) + + if is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token, + ) or is_seam_http_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ): + return get_auth_headers_for_console_session_token( + console_session_token, workspace_id + ) + + if is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + ) or is_seam_http_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + workspace_id=workspace_id, + ): + return get_auth_headers_for_personal_access_token( + personal_access_token, workspace_id + ) + + raise SeamHttpInvalidOptionsError( + "Must specify an api_key, client_session_token, publishable_key, console_session_token, " + "or personal_access_token. Attempted reading configuration from the environment, " + "but the environment variable SEAM_API_KEY is not set." + ) + + +def get_auth_headers_for_publishable_key(publishable_key: str) -> dict: + if is_jwt(publishable_key): + raise SeamHttpInvalidTokenError("A JWT cannot be used as a publishable_key") + + if is_access_token(publishable_key): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a publishable_key" + ) + + if is_client_session_token(publishable_key): + raise SeamHttpInvalidTokenError( + "A Client Session Token Key cannot be used as a publishable_key" + ) + + if not is_publishable_key(publishable_key): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid publishable_key format, expected token to start with {publishable_key_token_prefix}" + ) + + return {"seam-publishable-key": publishable_key} + + +def get_auth_headers_for_api_key(api_key: str) -> dict: + if is_client_session_token(api_key): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as an api_key" + ) + + if is_jwt(api_key): + raise SeamHttpInvalidTokenError("A JWT cannot be used as an api_key") + + if is_access_token(api_key): + raise SeamHttpInvalidTokenError("An Access Token cannot be used as an api_key") + + if is_publishable_key(api_key): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as an api_key" + ) + + if not is_seam_token(api_key): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid api_key format, expected token to start with {token_prefix}" + ) + + return {"authorization": f"Bearer {api_key}"} + + +def get_auth_headers_for_client_session_token(client_session_token: str) -> dict: + if is_jwt(client_session_token): + raise SeamHttpInvalidTokenError( + "A JWT cannot be used as a client_session_token" + ) + + if is_access_token(client_session_token): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a client_session_token" + ) + + if is_publishable_key(client_session_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a client_session_token" + ) + + if not is_client_session_token(client_session_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid client_session_token format, expected token to start with {client_session_token_prefix}" + ) + + return { + "authorization": f"Bearer {client_session_token}", + "client-session-token": client_session_token, + } + + +def get_auth_headers_for_console_session_token( + console_session_token: str, workspace_id: Optional[str] = None +) -> dict: + if is_access_token(console_session_token): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a console_session_token" + ) + + if is_client_session_token(console_session_token): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as a console_session_token" + ) + + if is_publishable_key(console_session_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a console_session_token" + ) + + if not is_jwt(console_session_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid console_session_token format, expected a JWT which starts with {jwt_prefix}" + ) + + headers = {"authorization": f"Bearer {console_session_token}"} + if workspace_id is not None: + headers["seam-workspace"] = workspace_id + + return headers + + +def get_auth_headers_for_personal_access_token( + personal_access_token: str, workspace_id: Optional[str] = None +) -> dict: + if is_jwt(personal_access_token): + raise SeamHttpInvalidTokenError( + "A JWT cannot be used as a personal_access_token" + ) + + if is_client_session_token(personal_access_token): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as a personal_access_token" + ) + + if is_publishable_key(personal_access_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a personal_access_token" + ) + + if not is_access_token(personal_access_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid personal_access_token format, expected token to start with {access_token_prefix}" + ) + + headers = {"authorization": f"Bearer {personal_access_token}"} + if workspace_id is not None: + headers["seam-workspace"] = workspace_id + + return headers + + +def warn_on_insecure_user_identifier_key(user_identifier_key: str): + if is_email(user_identifier_key): + warning_message = ( + "\033[93m" + "Using an email for the userIdentifierKey is insecure and may return an error in the future!\n" + "This is insecure because an email is common knowledge or easily guessed.\n" + "Use something with sufficient entropy known only to the owner of the client session.\n" + "For help choosing a user identifier key see " + "https://docs.seam.co/latest/seam-components/overview/get-started-with-client-side-components#3-select-a-user-identifier-key" + "\033[0m" + ) + print(warning_message) + + +def is_email(value: str) -> bool: + return re.match(r"^[^\s@]+@[^\s@]+\.[^\s@]+$", value) is not None From 42dc5069439b0b38bf2d9d680023a684c5b5ad08 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 21:58:02 +0200 Subject: [PATCH 02/21] Add options utils --- seam/utils/options.py | 155 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 seam/utils/options.py diff --git a/seam/utils/options.py b/seam/utils/options.py new file mode 100644 index 00000000..d70b82dd --- /dev/null +++ b/seam/utils/options.py @@ -0,0 +1,155 @@ +from typing import Optional + +class SeamHttpInvalidOptionsError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received invalid options: {message}") + +def is_seam_http_options_with_api_key( + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if api_key is None: + return False + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The client_session_token option cannot be used with the api_key option" + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The console_session_token option cannot be used with the api_key option" + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the api_key option" + ) + + return True + +def is_seam_http_options_with_client_session_token( + client_session_token: Optional[str] = None, + api_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None +) -> bool: + if client_session_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + 'The api_key option cannot be used with the client_session_token option' + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + 'The console_session_token option cannot be used with the client_session_token option' + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + 'The personal_access_token option cannot be used with the client_session_token option' + ) + + return True + +def is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None +) -> bool: + if console_session_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + 'The api_key option cannot be used with the console_session_token option' + ) + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + 'The client_session_token option cannot be used with the console_session_token option' + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + 'The personal_access_token option cannot be used with the console_session_token option' + ) + + return True + +def is_seam_http_options_with_console_session_token( + console_session_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None +) -> bool: + if not is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token + ): + return False + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + 'Must pass a workspace_id when using a console_session_token' + ) + + return True + +def is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None +) -> bool: + if personal_access_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + 'The api_key option cannot be used with the personal_access_token option' + ) + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + 'The client_session_token option cannot be used with the personal_access_token option' + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + 'The console_session_token option cannot be used with the personal_access_token option' + ) + + return True + +def is_seam_http_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + workspace_id: Optional[str] = None +) -> bool: + if not is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token + ): + return False + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + 'Must pass a workspace_id when using a personal_access_token' + ) + + return True + + From 02498979979b3f5374a7fc93045283698e15dc27 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 21:58:15 +0200 Subject: [PATCH 03/21] Add parse_options utils --- seam/utils/parse_options.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 seam/utils/parse_options.py diff --git a/seam/utils/parse_options.py b/seam/utils/parse_options.py new file mode 100644 index 00000000..ed8f086c --- /dev/null +++ b/seam/utils/parse_options.py @@ -0,0 +1,19 @@ +import os +from typing import Optional + + +def get_api_key_from_env( + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +): + if client_session_token is not None: + return None + + if console_session_token is not None: + return None + + if personal_access_token is not None: + return None + + return os.getenv("SEAM_API_KEY") From 36b11a9f11dbcdbd6131805171797a7b34e78679 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 21:58:25 +0200 Subject: [PATCH 04/21] Add token utils --- seam/utils/token.py | 47 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 seam/utils/token.py diff --git a/seam/utils/token.py b/seam/utils/token.py new file mode 100644 index 00000000..3c94c9f5 --- /dev/null +++ b/seam/utils/token.py @@ -0,0 +1,47 @@ +token_prefix = "seam_" + +access_token_prefix = "seam_at" + +jwt_prefix = "ey" + +client_session_token_prefix = "seam_cst" + +publishable_key_token_prefix = "seam_pk" + + +def is_access_token(token: str) -> bool: + return token.startswith(access_token_prefix) + + +def is_jwt(token: str) -> bool: + return token.startswith(jwt_prefix) + + +def is_seam_token(token: str) -> bool: + return token.startswith(token_prefix) + + +def is_api_key(token: str) -> bool: + return ( + not is_client_session_token(token) + and not is_jwt(token) + and not is_access_token(token) + and not is_publishable_key(token) + and is_seam_token(token) + ) + + +def is_client_session_token(token: str) -> bool: + return token.startswith(client_session_token_prefix) + + +def is_publishable_key(token: str) -> bool: + return token.startswith(publishable_key_token_prefix) + + +def is_console_session_token(token: str) -> bool: + return is_jwt(token) + + +def is_personal_access_token(token: str) -> bool: + return is_access_token(token) From f0d4de1c278395c94d65c1b8ca424cdf0b9025f0 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 21:59:08 +0200 Subject: [PATCH 05/21] Remove make_request from routes class --- seam/routes.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/seam/routes.py b/seam/routes.py index e5936481..35ba96ef 100644 --- a/seam/routes.py +++ b/seam/routes.py @@ -35,6 +35,3 @@ def __init__(self): self.user_identities = UserIdentities(seam=self) self.webhooks = Webhooks(seam=self) self.workspaces = Workspaces(seam=self) - - def make_request(self, method: str, path: str, **kwargs): - raise NotImplementedError() From 69037d8a68509276899c94f7f5235c1bbdabecbe Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:00:48 +0200 Subject: [PATCH 06/21] Update seam abstract class to include new auth methods --- seam/types.py | 89 ++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 78 insertions(+), 11 deletions(-) diff --git a/seam/types.py b/seam/types.py index 7adc75e0..ad374c0f 100644 --- a/seam/types.py +++ b/seam/types.py @@ -1,4 +1,5 @@ from typing import Any, Dict, List, Optional, Union +from typing_extensions import Self import abc from dataclasses import dataclass from seam.utils.deep_attr_dict import DeepAttrDict @@ -2061,26 +2062,92 @@ class AbstractRoutes(abc.ABC): webhooks: AbstractWebhooks workspaces: AbstractWorkspaces - @abc.abstractmethod - def make_request(self, method: str, path: str, **kwargs) -> Any: - raise NotImplementedError - -@dataclass class AbstractSeam(AbstractRoutes): - api_key: str - workspace_id: str - api_url: str lts_version: str - wait_for_action_attempt: bool @abc.abstractmethod def __init__( self, api_key: Optional[str] = None, *, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + user_identifier_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, - wait_for_action_attempt: Optional[bool] = False, + endpoint: Optional[str] = "https://connect.getseam.com", + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): + self.api_key = api_key + self.client_session_token = client_session_token + self.publishable_key = publishable_key + self.user_identifier_key = user_identifier_key + self.console_session_token = console_session_token + self.personal_access_token = personal_access_token + self.workspace_id = workspace_id + self.endpoint = endpoint + self.wait_for_action_attempt = wait_for_action_attempt + + @abc.abstractmethod + def make_request(self, method: str, path: str, **kwargs) -> Any: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_client_session_token( + cls, + client_session_token: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_publishable_key( + cls, + publishable_key: str, + user_identifier_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_console_session_token( + cls, + console_session_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: raise NotImplementedError From 88e05cd9e64473eea48c66e0ea21c600343bb706 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:01:08 +0200 Subject: [PATCH 07/21] Update seam client to support new auth methods --- seam/seam.py | 158 ++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 130 insertions(+), 28 deletions(-) diff --git a/seam/seam.py b/seam/seam.py index 1efc2153..3b0da4bd 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -1,9 +1,11 @@ import os - -from .routes import Routes import requests from importlib.metadata import version from typing import Optional, Union, Dict, cast +from typing_extensions import Self +from seam.utils.auth import get_auth_headers, warn_on_insecure_user_identifier_key +from seam.utils.parse_options import get_api_key_from_env +from .routes import Routes from .types import AbstractSeam, SeamApiException @@ -12,40 +14,58 @@ class Seam(AbstractSeam): Initial Seam class used to interact with Seam API """ - api_key: str - api_url: str = "https://connect.getseam.com" lts_version: str = "1.0.0" def __init__( self, api_key: Optional[str] = None, *, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, + endpoint: Optional[str] = "https://connect.getseam.com", wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): """ Parameters ---------- api_key : str, optional - API key + API key. + client_session_token : str, optional + Client session token. + publishable_key : str, optional + Publishable key. + user_identifier_key : str, optional + User identifier key. + console_session_token : str, optional + Console session token. + personal_access_token : str, optional + Personal access token. workspace_id : str, optional - Workspace id - api_url : str, optional - API url + Workspace id. + endpoint : str, optional + The API endpoint to which the request should be sent. Defaults to `https://connect.getseam.com`. + wait_for_action_attempt : bool or dict, optional + Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`. """ + Routes.__init__(self) - if api_key is None: - api_key = os.environ.get("SEAM_API_KEY", None) - if api_key is None: - raise Exception( - "SEAM_API_KEY not found in environment, and api_key not provided" - ) - if workspace_id is None: - workspace_id = os.environ.get("SEAM_WORKSPACE_ID", None) - self.api_key = api_key - self.workspace_id = workspace_id + api_key = api_key or get_api_key_from_env( + client_session_token=client_session_token, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ) + self.__auth_headers = get_auth_headers( + api_key=api_key, + client_session_token=client_session_token, + publishable_key=publishable_key, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ) self.lts_version = Seam.lts_version self.wait_for_action_attempt = wait_for_action_attempt @@ -57,13 +77,13 @@ def __init__( "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." "\033[0m" ) - api_url = ( + endpoint = ( os.environ.get("SEAM_API_URL", None) or os.environ.get("SEAM_ENDPOINT", None) - or api_url + or endpoint ) - if api_url is not None: - self.api_url = cast(str, api_url) + if endpoint is not None: + self.endpoint = cast(str, endpoint) def make_request(self, method: str, path: str, **kwargs): """ @@ -79,20 +99,18 @@ def make_request(self, method: str, path: str, **kwargs): Keyword arguments passed to requests.request """ - url = self.api_url + path + url = self.endpoint + path sdk_version = version("seam") headers = { - "Authorization": "Bearer " + self.api_key, + **self.__auth_headers, "Content-Type": "application/json", "User-Agent": "Python SDK v" + sdk_version - + " (https://github.com/seamapi/python)", + + " (https://github.com/seamapi/python-next)", "seam-sdk-name": "seamapi/python", "seam-sdk-version": sdk_version, "seam-lts-version": self.lts_version, } - if self.workspace_id is not None: - headers["seam-workspace"] = self.workspace_id response = requests.request(method, url, headers=headers, **kwargs) if response.status_code != 200: @@ -102,3 +120,87 @@ def make_request(self, method: str, path: str, **kwargs): return response.json() return response.text + + @classmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt + ) + + @classmethod + def from_client_session_token( + cls, + client_session_token: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + client_session_token=client_session_token, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_publishable_key( + cls, + publishable_key: str, + user_identifier_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + warn_on_insecure_user_identifier_key(publishable_key) + + seam = cls( + publishable_key=publishable_key, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + client_session = seam.client_sessions.get_or_create( + user_identifier_key=user_identifier_key + ) + + return cls.from_client_session_token( + client_session.token, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_console_session_token( + cls, + console_session_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + console_session_token=console_session_token, + workspace_id=workspace_id, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + personal_access_token=personal_access_token, + workspace_id=workspace_id, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) From 1a872f3a83eb8180dec601c24035b9dbb1554d19 Mon Sep 17 00:00:00 2001 From: Seam Bot Date: Thu, 25 Apr 2024 20:02:52 +0000 Subject: [PATCH 08/21] ci: Format code --- seam/utils/options.py | 45 ++++++++++++++++++++++++------------------- 1 file changed, 25 insertions(+), 20 deletions(-) diff --git a/seam/utils/options.py b/seam/utils/options.py index d70b82dd..f51085f8 100644 --- a/seam/utils/options.py +++ b/seam/utils/options.py @@ -1,9 +1,11 @@ from typing import Optional + class SeamHttpInvalidOptionsError(Exception): def __init__(self, message): super().__init__(f"SeamHttp received invalid options: {message}") + def is_seam_http_options_with_api_key( api_key: Optional[str] = None, client_session_token: Optional[str] = None, @@ -30,126 +32,129 @@ def is_seam_http_options_with_api_key( return True + def is_seam_http_options_with_client_session_token( client_session_token: Optional[str] = None, api_key: Optional[str] = None, console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None + personal_access_token: Optional[str] = None, ) -> bool: if client_session_token is None: return False if api_key is not None: raise SeamHttpInvalidOptionsError( - 'The api_key option cannot be used with the client_session_token option' + "The api_key option cannot be used with the client_session_token option" ) if console_session_token is not None: raise SeamHttpInvalidOptionsError( - 'The console_session_token option cannot be used with the client_session_token option' + "The console_session_token option cannot be used with the client_session_token option" ) if personal_access_token is not None: raise SeamHttpInvalidOptionsError( - 'The personal_access_token option cannot be used with the client_session_token option' + "The personal_access_token option cannot be used with the client_session_token option" ) return True + def is_seam_http_multi_workspace_options_with_console_session_token( console_session_token: Optional[str] = None, api_key: Optional[str] = None, client_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None + personal_access_token: Optional[str] = None, ) -> bool: if console_session_token is None: return False if api_key is not None: raise SeamHttpInvalidOptionsError( - 'The api_key option cannot be used with the console_session_token option' + "The api_key option cannot be used with the console_session_token option" ) if client_session_token is not None: raise SeamHttpInvalidOptionsError( - 'The client_session_token option cannot be used with the console_session_token option' + "The client_session_token option cannot be used with the console_session_token option" ) if personal_access_token is not None: raise SeamHttpInvalidOptionsError( - 'The personal_access_token option cannot be used with the console_session_token option' + "The personal_access_token option cannot be used with the console_session_token option" ) return True + def is_seam_http_options_with_console_session_token( console_session_token: Optional[str] = None, api_key: Optional[str] = None, client_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, - workspace_id: Optional[str] = None + workspace_id: Optional[str] = None, ) -> bool: if not is_seam_http_multi_workspace_options_with_console_session_token( console_session_token=console_session_token, api_key=api_key, client_session_token=client_session_token, - personal_access_token=personal_access_token + personal_access_token=personal_access_token, ): return False if workspace_id is None: raise SeamHttpInvalidOptionsError( - 'Must pass a workspace_id when using a console_session_token' + "Must pass a workspace_id when using a console_session_token" ) return True + def is_seam_http_multi_workspace_options_with_personal_access_token( personal_access_token: Optional[str] = None, api_key: Optional[str] = None, client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None + console_session_token: Optional[str] = None, ) -> bool: if personal_access_token is None: return False if api_key is not None: raise SeamHttpInvalidOptionsError( - 'The api_key option cannot be used with the personal_access_token option' + "The api_key option cannot be used with the personal_access_token option" ) if client_session_token is not None: raise SeamHttpInvalidOptionsError( - 'The client_session_token option cannot be used with the personal_access_token option' + "The client_session_token option cannot be used with the personal_access_token option" ) if console_session_token is not None: raise SeamHttpInvalidOptionsError( - 'The console_session_token option cannot be used with the personal_access_token option' + "The console_session_token option cannot be used with the personal_access_token option" ) return True + def is_seam_http_options_with_personal_access_token( personal_access_token: Optional[str] = None, api_key: Optional[str] = None, client_session_token: Optional[str] = None, console_session_token: Optional[str] = None, - workspace_id: Optional[str] = None + workspace_id: Optional[str] = None, ) -> bool: if not is_seam_http_multi_workspace_options_with_personal_access_token( personal_access_token=personal_access_token, api_key=api_key, client_session_token=client_session_token, - console_session_token=console_session_token + console_session_token=console_session_token, ): return False if workspace_id is None: raise SeamHttpInvalidOptionsError( - 'Must pass a workspace_id when using a personal_access_token' + "Must pass a workspace_id when using a personal_access_token" ) return True - - From 77209fb77c44c65f5553a952beaca2ff511afce8 Mon Sep 17 00:00:00 2001 From: Seam Bot Date: Thu, 25 Apr 2024 20:03:29 +0000 Subject: [PATCH 09/21] ci: Generate code --- seam/routes.py | 3 + seam/seam.py | 158 ++++------------------- seam/types.py | 89 ++----------- seam/utils/auth.py | 246 ------------------------------------ seam/utils/options.py | 160 ----------------------- seam/utils/parse_options.py | 19 --- seam/utils/token.py | 47 ------- 7 files changed, 42 insertions(+), 680 deletions(-) delete mode 100644 seam/utils/auth.py delete mode 100644 seam/utils/options.py delete mode 100644 seam/utils/parse_options.py delete mode 100644 seam/utils/token.py diff --git a/seam/routes.py b/seam/routes.py index 35ba96ef..e5936481 100644 --- a/seam/routes.py +++ b/seam/routes.py @@ -35,3 +35,6 @@ def __init__(self): self.user_identities = UserIdentities(seam=self) self.webhooks = Webhooks(seam=self) self.workspaces = Workspaces(seam=self) + + def make_request(self, method: str, path: str, **kwargs): + raise NotImplementedError() diff --git a/seam/seam.py b/seam/seam.py index 3b0da4bd..1efc2153 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -1,11 +1,9 @@ import os + +from .routes import Routes import requests from importlib.metadata import version from typing import Optional, Union, Dict, cast -from typing_extensions import Self -from seam.utils.auth import get_auth_headers, warn_on_insecure_user_identifier_key -from seam.utils.parse_options import get_api_key_from_env -from .routes import Routes from .types import AbstractSeam, SeamApiException @@ -14,58 +12,40 @@ class Seam(AbstractSeam): Initial Seam class used to interact with Seam API """ + api_key: str + api_url: str = "https://connect.getseam.com" lts_version: str = "1.0.0" def __init__( self, api_key: Optional[str] = None, *, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - endpoint: Optional[str] = "https://connect.getseam.com", + api_url: Optional[str] = None, wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): """ Parameters ---------- api_key : str, optional - API key. - client_session_token : str, optional - Client session token. - publishable_key : str, optional - Publishable key. - user_identifier_key : str, optional - User identifier key. - console_session_token : str, optional - Console session token. - personal_access_token : str, optional - Personal access token. + API key workspace_id : str, optional - Workspace id. - endpoint : str, optional - The API endpoint to which the request should be sent. Defaults to `https://connect.getseam.com`. - wait_for_action_attempt : bool or dict, optional - Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`. + Workspace id + api_url : str, optional + API url """ - Routes.__init__(self) - api_key = api_key or get_api_key_from_env( - client_session_token=client_session_token, - console_session_token=console_session_token, - personal_access_token=personal_access_token, - ) - self.__auth_headers = get_auth_headers( - api_key=api_key, - client_session_token=client_session_token, - publishable_key=publishable_key, - console_session_token=console_session_token, - personal_access_token=personal_access_token, - workspace_id=workspace_id, - ) + if api_key is None: + api_key = os.environ.get("SEAM_API_KEY", None) + if api_key is None: + raise Exception( + "SEAM_API_KEY not found in environment, and api_key not provided" + ) + if workspace_id is None: + workspace_id = os.environ.get("SEAM_WORKSPACE_ID", None) + self.api_key = api_key + self.workspace_id = workspace_id self.lts_version = Seam.lts_version self.wait_for_action_attempt = wait_for_action_attempt @@ -77,13 +57,13 @@ def __init__( "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." "\033[0m" ) - endpoint = ( + api_url = ( os.environ.get("SEAM_API_URL", None) or os.environ.get("SEAM_ENDPOINT", None) - or endpoint + or api_url ) - if endpoint is not None: - self.endpoint = cast(str, endpoint) + if api_url is not None: + self.api_url = cast(str, api_url) def make_request(self, method: str, path: str, **kwargs): """ @@ -99,18 +79,20 @@ def make_request(self, method: str, path: str, **kwargs): Keyword arguments passed to requests.request """ - url = self.endpoint + path + url = self.api_url + path sdk_version = version("seam") headers = { - **self.__auth_headers, + "Authorization": "Bearer " + self.api_key, "Content-Type": "application/json", "User-Agent": "Python SDK v" + sdk_version - + " (https://github.com/seamapi/python-next)", + + " (https://github.com/seamapi/python)", "seam-sdk-name": "seamapi/python", "seam-sdk-version": sdk_version, "seam-lts-version": self.lts_version, } + if self.workspace_id is not None: + headers["seam-workspace"] = self.workspace_id response = requests.request(method, url, headers=headers, **kwargs) if response.status_code != 200: @@ -120,87 +102,3 @@ def make_request(self, method: str, path: str, **kwargs): return response.json() return response.text - - @classmethod - def from_api_key( - cls, - api_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt - ) - - @classmethod - def from_client_session_token( - cls, - client_session_token: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - client_session_token=client_session_token, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - - @classmethod - def from_publishable_key( - cls, - publishable_key: str, - user_identifier_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - warn_on_insecure_user_identifier_key(publishable_key) - - seam = cls( - publishable_key=publishable_key, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - client_session = seam.client_sessions.get_or_create( - user_identifier_key=user_identifier_key - ) - - return cls.from_client_session_token( - client_session.token, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - - @classmethod - def from_console_session_token( - cls, - console_session_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - console_session_token=console_session_token, - workspace_id=workspace_id, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - - @classmethod - def from_personal_access_token( - cls, - personal_access_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - personal_access_token=personal_access_token, - workspace_id=workspace_id, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) diff --git a/seam/types.py b/seam/types.py index ad374c0f..7adc75e0 100644 --- a/seam/types.py +++ b/seam/types.py @@ -1,5 +1,4 @@ from typing import Any, Dict, List, Optional, Union -from typing_extensions import Self import abc from dataclasses import dataclass from seam.utils.deep_attr_dict import DeepAttrDict @@ -2062,92 +2061,26 @@ class AbstractRoutes(abc.ABC): webhooks: AbstractWebhooks workspaces: AbstractWorkspaces + @abc.abstractmethod + def make_request(self, method: str, path: str, **kwargs) -> Any: + raise NotImplementedError + +@dataclass class AbstractSeam(AbstractRoutes): + api_key: str + workspace_id: str + api_url: str lts_version: str + wait_for_action_attempt: bool @abc.abstractmethod def __init__( self, api_key: Optional[str] = None, *, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - user_identifier_key: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - endpoint: Optional[str] = "https://connect.getseam.com", - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + api_url: Optional[str] = None, + wait_for_action_attempt: Optional[bool] = False, ): - self.api_key = api_key - self.client_session_token = client_session_token - self.publishable_key = publishable_key - self.user_identifier_key = user_identifier_key - self.console_session_token = console_session_token - self.personal_access_token = personal_access_token - self.workspace_id = workspace_id - self.endpoint = endpoint - self.wait_for_action_attempt = wait_for_action_attempt - - @abc.abstractmethod - def make_request(self, method: str, path: str, **kwargs) -> Any: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_api_key( - cls, - api_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_client_session_token( - cls, - client_session_token: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_publishable_key( - cls, - publishable_key: str, - user_identifier_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_console_session_token( - cls, - console_session_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_personal_access_token( - cls, - personal_access_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: raise NotImplementedError diff --git a/seam/utils/auth.py b/seam/utils/auth.py deleted file mode 100644 index fa6337cb..00000000 --- a/seam/utils/auth.py +++ /dev/null @@ -1,246 +0,0 @@ -from typing import Optional -import re -from seam.utils.options import ( - SeamHttpInvalidOptionsError, - is_seam_http_multi_workspace_options_with_personal_access_token, - is_seam_http_options_with_api_key, - is_seam_http_options_with_client_session_token, - is_seam_http_multi_workspace_options_with_console_session_token, - is_seam_http_options_with_console_session_token, - is_seam_http_options_with_personal_access_token, -) -from seam.utils.token import ( - is_jwt, - is_access_token, - is_client_session_token, - is_publishable_key, - is_seam_token, - publishable_key_token_prefix, - token_prefix, - client_session_token_prefix, - jwt_prefix, - access_token_prefix, -) - - -class SeamHttpInvalidTokenError(Exception): - def __init__(self, message): - super().__init__(f"SeamHttp received an invalid token: {message}") - - -def get_auth_headers( - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, - workspace_id: Optional[str] = None, -): - if publishable_key: - return get_auth_headers_for_publishable_key(publishable_key) - - if is_seam_http_options_with_api_key( - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - personal_access_token=personal_access_token, - ): - return get_auth_headers_for_api_key(api_key) - - if is_seam_http_options_with_client_session_token( - client_session_token=client_session_token, - api_key=api_key, - console_session_token=console_session_token, - personal_access_token=personal_access_token, - ): - return get_auth_headers_for_client_session_token(client_session_token) - - if is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token=console_session_token, - api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - ) or is_seam_http_options_with_console_session_token( - console_session_token=console_session_token, - api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - workspace_id=workspace_id, - ): - return get_auth_headers_for_console_session_token( - console_session_token, workspace_id - ) - - if is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - ) or is_seam_http_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - workspace_id=workspace_id, - ): - return get_auth_headers_for_personal_access_token( - personal_access_token, workspace_id - ) - - raise SeamHttpInvalidOptionsError( - "Must specify an api_key, client_session_token, publishable_key, console_session_token, " - "or personal_access_token. Attempted reading configuration from the environment, " - "but the environment variable SEAM_API_KEY is not set." - ) - - -def get_auth_headers_for_publishable_key(publishable_key: str) -> dict: - if is_jwt(publishable_key): - raise SeamHttpInvalidTokenError("A JWT cannot be used as a publishable_key") - - if is_access_token(publishable_key): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a publishable_key" - ) - - if is_client_session_token(publishable_key): - raise SeamHttpInvalidTokenError( - "A Client Session Token Key cannot be used as a publishable_key" - ) - - if not is_publishable_key(publishable_key): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid publishable_key format, expected token to start with {publishable_key_token_prefix}" - ) - - return {"seam-publishable-key": publishable_key} - - -def get_auth_headers_for_api_key(api_key: str) -> dict: - if is_client_session_token(api_key): - raise SeamHttpInvalidTokenError( - "A Client Session Token cannot be used as an api_key" - ) - - if is_jwt(api_key): - raise SeamHttpInvalidTokenError("A JWT cannot be used as an api_key") - - if is_access_token(api_key): - raise SeamHttpInvalidTokenError("An Access Token cannot be used as an api_key") - - if is_publishable_key(api_key): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as an api_key" - ) - - if not is_seam_token(api_key): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid api_key format, expected token to start with {token_prefix}" - ) - - return {"authorization": f"Bearer {api_key}"} - - -def get_auth_headers_for_client_session_token(client_session_token: str) -> dict: - if is_jwt(client_session_token): - raise SeamHttpInvalidTokenError( - "A JWT cannot be used as a client_session_token" - ) - - if is_access_token(client_session_token): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a client_session_token" - ) - - if is_publishable_key(client_session_token): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as a client_session_token" - ) - - if not is_client_session_token(client_session_token): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid client_session_token format, expected token to start with {client_session_token_prefix}" - ) - - return { - "authorization": f"Bearer {client_session_token}", - "client-session-token": client_session_token, - } - - -def get_auth_headers_for_console_session_token( - console_session_token: str, workspace_id: Optional[str] = None -) -> dict: - if is_access_token(console_session_token): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a console_session_token" - ) - - if is_client_session_token(console_session_token): - raise SeamHttpInvalidTokenError( - "A Client Session Token cannot be used as a console_session_token" - ) - - if is_publishable_key(console_session_token): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as a console_session_token" - ) - - if not is_jwt(console_session_token): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid console_session_token format, expected a JWT which starts with {jwt_prefix}" - ) - - headers = {"authorization": f"Bearer {console_session_token}"} - if workspace_id is not None: - headers["seam-workspace"] = workspace_id - - return headers - - -def get_auth_headers_for_personal_access_token( - personal_access_token: str, workspace_id: Optional[str] = None -) -> dict: - if is_jwt(personal_access_token): - raise SeamHttpInvalidTokenError( - "A JWT cannot be used as a personal_access_token" - ) - - if is_client_session_token(personal_access_token): - raise SeamHttpInvalidTokenError( - "A Client Session Token cannot be used as a personal_access_token" - ) - - if is_publishable_key(personal_access_token): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as a personal_access_token" - ) - - if not is_access_token(personal_access_token): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid personal_access_token format, expected token to start with {access_token_prefix}" - ) - - headers = {"authorization": f"Bearer {personal_access_token}"} - if workspace_id is not None: - headers["seam-workspace"] = workspace_id - - return headers - - -def warn_on_insecure_user_identifier_key(user_identifier_key: str): - if is_email(user_identifier_key): - warning_message = ( - "\033[93m" - "Using an email for the userIdentifierKey is insecure and may return an error in the future!\n" - "This is insecure because an email is common knowledge or easily guessed.\n" - "Use something with sufficient entropy known only to the owner of the client session.\n" - "For help choosing a user identifier key see " - "https://docs.seam.co/latest/seam-components/overview/get-started-with-client-side-components#3-select-a-user-identifier-key" - "\033[0m" - ) - print(warning_message) - - -def is_email(value: str) -> bool: - return re.match(r"^[^\s@]+@[^\s@]+\.[^\s@]+$", value) is not None diff --git a/seam/utils/options.py b/seam/utils/options.py deleted file mode 100644 index f51085f8..00000000 --- a/seam/utils/options.py +++ /dev/null @@ -1,160 +0,0 @@ -from typing import Optional - - -class SeamHttpInvalidOptionsError(Exception): - def __init__(self, message): - super().__init__(f"SeamHttp received invalid options: {message}") - - -def is_seam_http_options_with_api_key( - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, -) -> bool: - if api_key is None: - return False - - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the api_key option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the api_key option" - ) - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the api_key option" - ) - - return True - - -def is_seam_http_options_with_client_session_token( - client_session_token: Optional[str] = None, - api_key: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, -) -> bool: - if client_session_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the client_session_token option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the client_session_token option" - ) - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the client_session_token option" - ) - - return True - - -def is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, -) -> bool: - if console_session_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the console_session_token option" - ) - - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the console_session_token option" - ) - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the console_session_token option" - ) - - return True - - -def is_seam_http_options_with_console_session_token( - console_session_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, - workspace_id: Optional[str] = None, -) -> bool: - if not is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token=console_session_token, - api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - ): - return False - - if workspace_id is None: - raise SeamHttpInvalidOptionsError( - "Must pass a workspace_id when using a console_session_token" - ) - - return True - - -def is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, -) -> bool: - if personal_access_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the personal_access_token option" - ) - - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the personal_access_token option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the personal_access_token option" - ) - - return True - - -def is_seam_http_options_with_personal_access_token( - personal_access_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, - workspace_id: Optional[str] = None, -) -> bool: - if not is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - ): - return False - - if workspace_id is None: - raise SeamHttpInvalidOptionsError( - "Must pass a workspace_id when using a personal_access_token" - ) - - return True diff --git a/seam/utils/parse_options.py b/seam/utils/parse_options.py deleted file mode 100644 index ed8f086c..00000000 --- a/seam/utils/parse_options.py +++ /dev/null @@ -1,19 +0,0 @@ -import os -from typing import Optional - - -def get_api_key_from_env( - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, -): - if client_session_token is not None: - return None - - if console_session_token is not None: - return None - - if personal_access_token is not None: - return None - - return os.getenv("SEAM_API_KEY") diff --git a/seam/utils/token.py b/seam/utils/token.py deleted file mode 100644 index 3c94c9f5..00000000 --- a/seam/utils/token.py +++ /dev/null @@ -1,47 +0,0 @@ -token_prefix = "seam_" - -access_token_prefix = "seam_at" - -jwt_prefix = "ey" - -client_session_token_prefix = "seam_cst" - -publishable_key_token_prefix = "seam_pk" - - -def is_access_token(token: str) -> bool: - return token.startswith(access_token_prefix) - - -def is_jwt(token: str) -> bool: - return token.startswith(jwt_prefix) - - -def is_seam_token(token: str) -> bool: - return token.startswith(token_prefix) - - -def is_api_key(token: str) -> bool: - return ( - not is_client_session_token(token) - and not is_jwt(token) - and not is_access_token(token) - and not is_publishable_key(token) - and is_seam_token(token) - ) - - -def is_client_session_token(token: str) -> bool: - return token.startswith(client_session_token_prefix) - - -def is_publishable_key(token: str) -> bool: - return token.startswith(publishable_key_token_prefix) - - -def is_console_session_token(token: str) -> bool: - return is_jwt(token) - - -def is_personal_access_token(token: str) -> bool: - return is_access_token(token) From a98884dccfe51139394a15b288ee5371e1a3b428 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:28:26 +0200 Subject: [PATCH 10/21] Update how endpoint is determined --- .github/workflows/generate.yml | 2 +- seam/seam.py | 27 +++++++++++++--------- seam/utils/parse_options.py | 42 ++++++++++++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 12 deletions(-) create mode 100644 seam/utils/parse_options.py diff --git a/.github/workflows/generate.yml b/.github/workflows/generate.yml index 9fc56542..780a2612 100644 --- a/.github/workflows/generate.yml +++ b/.github/workflows/generate.yml @@ -4,7 +4,7 @@ name: Generate on: push: branches-ignore: - - main + - ** workflow_dispatch: {} jobs: diff --git a/seam/seam.py b/seam/seam.py index 1efc2153..fd762bc8 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -4,9 +4,16 @@ import requests from importlib.metadata import version from typing import Optional, Union, Dict, cast +from typing_extensions import Self +from seam.utils.auth import get_auth_headers, warn_on_insecure_user_identifier_key +from seam.utils.parse_options import get_api_key_from_env, get_endpoint_from_env +from .routes import Routes from .types import AbstractSeam, SeamApiException +DEFAULT_ENDPOINT = "https://connect.getseam.com" + + class Seam(AbstractSeam): """ Initial Seam class used to interact with Seam API @@ -21,7 +28,7 @@ def __init__( api_key: Optional[str] = None, *, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, + endpoint: Optional[str] = None, wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): """ @@ -30,9 +37,11 @@ def __init__( api_key : str, optional API key workspace_id : str, optional - Workspace id - api_url : str, optional - API url + Workspace id. + endpoint : str, optional + The API endpoint to which the request should be sent. + wait_for_action_attempt : bool or dict, optional + Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`. """ Routes.__init__(self) @@ -57,13 +66,9 @@ def __init__( "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." "\033[0m" ) - api_url = ( - os.environ.get("SEAM_API_URL", None) - or os.environ.get("SEAM_ENDPOINT", None) - or api_url - ) - if api_url is not None: - self.api_url = cast(str, api_url) + endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT + if endpoint is not None: + self.endpoint = cast(str, endpoint) def make_request(self, method: str, path: str, **kwargs): """ diff --git a/seam/utils/parse_options.py b/seam/utils/parse_options.py new file mode 100644 index 00000000..4f765e93 --- /dev/null +++ b/seam/utils/parse_options.py @@ -0,0 +1,42 @@ +import os +from typing import Optional + + +def get_api_key_from_env( + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +): + if client_session_token is not None: + return None + + if console_session_token is not None: + return None + + if personal_access_token is not None: + return None + + return os.getenv("SEAM_API_KEY") + + +def get_endpoint_from_env(): + seam_api_url = os.getenv("SEAM_API_URL") + seam_endpoint = os.getenv("SEAM_ENDPOINT") + + if seam_api_url is not None: + print( + "\033[93m" + "Using the SEAM_API_URL environment variable is deprecated. " + "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." + "\033[0m" + ) + + if seam_api_url is not None and seam_endpoint is not None: + print( + "\033[93m" + "Detected both the SEAM_API_URL and SEAM_ENDPOINT environment variables. " + "Using SEAM_ENDPOINT." + "\033[0m" + ) + + return seam_endpoint or seam_api_url From 594f47b3d8055dbbdb521c0121793cb165a1a47e Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:30:25 +0200 Subject: [PATCH 11/21] Revert "ci: Generate code" This reverts commit 77209fb77c44c65f5553a952beaca2ff511afce8. --- seam/routes.py | 3 - seam/seam.py | 136 +++++++++++++++++++---- seam/types.py | 89 +++++++++++++-- seam/utils/auth.py | 246 ++++++++++++++++++++++++++++++++++++++++++ seam/utils/options.py | 160 +++++++++++++++++++++++++++ seam/utils/token.py | 47 ++++++++ 6 files changed, 647 insertions(+), 34 deletions(-) create mode 100644 seam/utils/auth.py create mode 100644 seam/utils/options.py create mode 100644 seam/utils/token.py diff --git a/seam/routes.py b/seam/routes.py index e5936481..35ba96ef 100644 --- a/seam/routes.py +++ b/seam/routes.py @@ -35,6 +35,3 @@ def __init__(self): self.user_identities = UserIdentities(seam=self) self.webhooks = Webhooks(seam=self) self.workspaces = Workspaces(seam=self) - - def make_request(self, method: str, path: str, **kwargs): - raise NotImplementedError() diff --git a/seam/seam.py b/seam/seam.py index fd762bc8..44849a34 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -1,6 +1,4 @@ import os - -from .routes import Routes import requests from importlib.metadata import version from typing import Optional, Union, Dict, cast @@ -19,14 +17,16 @@ class Seam(AbstractSeam): Initial Seam class used to interact with Seam API """ - api_key: str - api_url: str = "https://connect.getseam.com" lts_version: str = "1.0.0" def __init__( self, api_key: Optional[str] = None, *, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, endpoint: Optional[str] = None, wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, @@ -35,7 +35,17 @@ def __init__( Parameters ---------- api_key : str, optional - API key + API key. + client_session_token : str, optional + Client session token. + publishable_key : str, optional + Publishable key. + user_identifier_key : str, optional + User identifier key. + console_session_token : str, optional + Console session token. + personal_access_token : str, optional + Personal access token. workspace_id : str, optional Workspace id. endpoint : str, optional @@ -43,18 +53,22 @@ def __init__( wait_for_action_attempt : bool or dict, optional Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`. """ + Routes.__init__(self) - if api_key is None: - api_key = os.environ.get("SEAM_API_KEY", None) - if api_key is None: - raise Exception( - "SEAM_API_KEY not found in environment, and api_key not provided" - ) - if workspace_id is None: - workspace_id = os.environ.get("SEAM_WORKSPACE_ID", None) - self.api_key = api_key - self.workspace_id = workspace_id + api_key = api_key or get_api_key_from_env( + client_session_token=client_session_token, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ) + self.__auth_headers = get_auth_headers( + api_key=api_key, + client_session_token=client_session_token, + publishable_key=publishable_key, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ) self.lts_version = Seam.lts_version self.wait_for_action_attempt = wait_for_action_attempt @@ -84,20 +98,18 @@ def make_request(self, method: str, path: str, **kwargs): Keyword arguments passed to requests.request """ - url = self.api_url + path + url = self.endpoint + path sdk_version = version("seam") headers = { - "Authorization": "Bearer " + self.api_key, + **self.__auth_headers, "Content-Type": "application/json", "User-Agent": "Python SDK v" + sdk_version - + " (https://github.com/seamapi/python)", + + " (https://github.com/seamapi/python-next)", "seam-sdk-name": "seamapi/python", "seam-sdk-version": sdk_version, "seam-lts-version": self.lts_version, } - if self.workspace_id is not None: - headers["seam-workspace"] = self.workspace_id response = requests.request(method, url, headers=headers, **kwargs) if response.status_code != 200: @@ -107,3 +119,87 @@ def make_request(self, method: str, path: str, **kwargs): return response.json() return response.text + + @classmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt + ) + + @classmethod + def from_client_session_token( + cls, + client_session_token: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + client_session_token=client_session_token, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_publishable_key( + cls, + publishable_key: str, + user_identifier_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + warn_on_insecure_user_identifier_key(publishable_key) + + seam = cls( + publishable_key=publishable_key, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + client_session = seam.client_sessions.get_or_create( + user_identifier_key=user_identifier_key + ) + + return cls.from_client_session_token( + client_session.token, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_console_session_token( + cls, + console_session_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + console_session_token=console_session_token, + workspace_id=workspace_id, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) + + @classmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + personal_access_token=personal_access_token, + workspace_id=workspace_id, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) diff --git a/seam/types.py b/seam/types.py index 7adc75e0..ad374c0f 100644 --- a/seam/types.py +++ b/seam/types.py @@ -1,4 +1,5 @@ from typing import Any, Dict, List, Optional, Union +from typing_extensions import Self import abc from dataclasses import dataclass from seam.utils.deep_attr_dict import DeepAttrDict @@ -2061,26 +2062,92 @@ class AbstractRoutes(abc.ABC): webhooks: AbstractWebhooks workspaces: AbstractWorkspaces - @abc.abstractmethod - def make_request(self, method: str, path: str, **kwargs) -> Any: - raise NotImplementedError - -@dataclass class AbstractSeam(AbstractRoutes): - api_key: str - workspace_id: str - api_url: str lts_version: str - wait_for_action_attempt: bool @abc.abstractmethod def __init__( self, api_key: Optional[str] = None, *, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + user_identifier_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, - wait_for_action_attempt: Optional[bool] = False, + endpoint: Optional[str] = "https://connect.getseam.com", + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): + self.api_key = api_key + self.client_session_token = client_session_token + self.publishable_key = publishable_key + self.user_identifier_key = user_identifier_key + self.console_session_token = console_session_token + self.personal_access_token = personal_access_token + self.workspace_id = workspace_id + self.endpoint = endpoint + self.wait_for_action_attempt = wait_for_action_attempt + + @abc.abstractmethod + def make_request(self, method: str, path: str, **kwargs) -> Any: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_client_session_token( + cls, + client_session_token: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_publishable_key( + cls, + publishable_key: str, + user_identifier_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_console_session_token( + cls, + console_session_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: raise NotImplementedError diff --git a/seam/utils/auth.py b/seam/utils/auth.py new file mode 100644 index 00000000..fa6337cb --- /dev/null +++ b/seam/utils/auth.py @@ -0,0 +1,246 @@ +from typing import Optional +import re +from seam.utils.options import ( + SeamHttpInvalidOptionsError, + is_seam_http_multi_workspace_options_with_personal_access_token, + is_seam_http_options_with_api_key, + is_seam_http_options_with_client_session_token, + is_seam_http_multi_workspace_options_with_console_session_token, + is_seam_http_options_with_console_session_token, + is_seam_http_options_with_personal_access_token, +) +from seam.utils.token import ( + is_jwt, + is_access_token, + is_client_session_token, + is_publishable_key, + is_seam_token, + publishable_key_token_prefix, + token_prefix, + client_session_token_prefix, + jwt_prefix, + access_token_prefix, +) + + +class SeamHttpInvalidTokenError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received an invalid token: {message}") + + +def get_auth_headers( + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + publishable_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, +): + if publishable_key: + return get_auth_headers_for_publishable_key(publishable_key) + + if is_seam_http_options_with_api_key( + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ): + return get_auth_headers_for_api_key(api_key) + + if is_seam_http_options_with_client_session_token( + client_session_token=client_session_token, + api_key=api_key, + console_session_token=console_session_token, + personal_access_token=personal_access_token, + ): + return get_auth_headers_for_client_session_token(client_session_token) + + if is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token, + ) or is_seam_http_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ): + return get_auth_headers_for_console_session_token( + console_session_token, workspace_id + ) + + if is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + ) or is_seam_http_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + workspace_id=workspace_id, + ): + return get_auth_headers_for_personal_access_token( + personal_access_token, workspace_id + ) + + raise SeamHttpInvalidOptionsError( + "Must specify an api_key, client_session_token, publishable_key, console_session_token, " + "or personal_access_token. Attempted reading configuration from the environment, " + "but the environment variable SEAM_API_KEY is not set." + ) + + +def get_auth_headers_for_publishable_key(publishable_key: str) -> dict: + if is_jwt(publishable_key): + raise SeamHttpInvalidTokenError("A JWT cannot be used as a publishable_key") + + if is_access_token(publishable_key): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a publishable_key" + ) + + if is_client_session_token(publishable_key): + raise SeamHttpInvalidTokenError( + "A Client Session Token Key cannot be used as a publishable_key" + ) + + if not is_publishable_key(publishable_key): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid publishable_key format, expected token to start with {publishable_key_token_prefix}" + ) + + return {"seam-publishable-key": publishable_key} + + +def get_auth_headers_for_api_key(api_key: str) -> dict: + if is_client_session_token(api_key): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as an api_key" + ) + + if is_jwt(api_key): + raise SeamHttpInvalidTokenError("A JWT cannot be used as an api_key") + + if is_access_token(api_key): + raise SeamHttpInvalidTokenError("An Access Token cannot be used as an api_key") + + if is_publishable_key(api_key): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as an api_key" + ) + + if not is_seam_token(api_key): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid api_key format, expected token to start with {token_prefix}" + ) + + return {"authorization": f"Bearer {api_key}"} + + +def get_auth_headers_for_client_session_token(client_session_token: str) -> dict: + if is_jwt(client_session_token): + raise SeamHttpInvalidTokenError( + "A JWT cannot be used as a client_session_token" + ) + + if is_access_token(client_session_token): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a client_session_token" + ) + + if is_publishable_key(client_session_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a client_session_token" + ) + + if not is_client_session_token(client_session_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid client_session_token format, expected token to start with {client_session_token_prefix}" + ) + + return { + "authorization": f"Bearer {client_session_token}", + "client-session-token": client_session_token, + } + + +def get_auth_headers_for_console_session_token( + console_session_token: str, workspace_id: Optional[str] = None +) -> dict: + if is_access_token(console_session_token): + raise SeamHttpInvalidTokenError( + "An Access Token cannot be used as a console_session_token" + ) + + if is_client_session_token(console_session_token): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as a console_session_token" + ) + + if is_publishable_key(console_session_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a console_session_token" + ) + + if not is_jwt(console_session_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid console_session_token format, expected a JWT which starts with {jwt_prefix}" + ) + + headers = {"authorization": f"Bearer {console_session_token}"} + if workspace_id is not None: + headers["seam-workspace"] = workspace_id + + return headers + + +def get_auth_headers_for_personal_access_token( + personal_access_token: str, workspace_id: Optional[str] = None +) -> dict: + if is_jwt(personal_access_token): + raise SeamHttpInvalidTokenError( + "A JWT cannot be used as a personal_access_token" + ) + + if is_client_session_token(personal_access_token): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as a personal_access_token" + ) + + if is_publishable_key(personal_access_token): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as a personal_access_token" + ) + + if not is_access_token(personal_access_token): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid personal_access_token format, expected token to start with {access_token_prefix}" + ) + + headers = {"authorization": f"Bearer {personal_access_token}"} + if workspace_id is not None: + headers["seam-workspace"] = workspace_id + + return headers + + +def warn_on_insecure_user_identifier_key(user_identifier_key: str): + if is_email(user_identifier_key): + warning_message = ( + "\033[93m" + "Using an email for the userIdentifierKey is insecure and may return an error in the future!\n" + "This is insecure because an email is common knowledge or easily guessed.\n" + "Use something with sufficient entropy known only to the owner of the client session.\n" + "For help choosing a user identifier key see " + "https://docs.seam.co/latest/seam-components/overview/get-started-with-client-side-components#3-select-a-user-identifier-key" + "\033[0m" + ) + print(warning_message) + + +def is_email(value: str) -> bool: + return re.match(r"^[^\s@]+@[^\s@]+\.[^\s@]+$", value) is not None diff --git a/seam/utils/options.py b/seam/utils/options.py new file mode 100644 index 00000000..f51085f8 --- /dev/null +++ b/seam/utils/options.py @@ -0,0 +1,160 @@ +from typing import Optional + + +class SeamHttpInvalidOptionsError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received invalid options: {message}") + + +def is_seam_http_options_with_api_key( + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if api_key is None: + return False + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The client_session_token option cannot be used with the api_key option" + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The console_session_token option cannot be used with the api_key option" + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the api_key option" + ) + + return True + + +def is_seam_http_options_with_client_session_token( + client_session_token: Optional[str] = None, + api_key: Optional[str] = None, + console_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if client_session_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + "The api_key option cannot be used with the client_session_token option" + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The console_session_token option cannot be used with the client_session_token option" + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the client_session_token option" + ) + + return True + + +def is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if console_session_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + "The api_key option cannot be used with the console_session_token option" + ) + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The client_session_token option cannot be used with the console_session_token option" + ) + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the console_session_token option" + ) + + return True + + +def is_seam_http_options_with_console_session_token( + console_session_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, +) -> bool: + if not is_seam_http_multi_workspace_options_with_console_session_token( + console_session_token=console_session_token, + api_key=api_key, + client_session_token=client_session_token, + personal_access_token=personal_access_token, + ): + return False + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + "Must pass a workspace_id when using a console_session_token" + ) + + return True + + +def is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, +) -> bool: + if personal_access_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + "The api_key option cannot be used with the personal_access_token option" + ) + + if client_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The client_session_token option cannot be used with the personal_access_token option" + ) + + if console_session_token is not None: + raise SeamHttpInvalidOptionsError( + "The console_session_token option cannot be used with the personal_access_token option" + ) + + return True + + +def is_seam_http_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + client_session_token: Optional[str] = None, + console_session_token: Optional[str] = None, + workspace_id: Optional[str] = None, +) -> bool: + if not is_seam_http_multi_workspace_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + client_session_token=client_session_token, + console_session_token=console_session_token, + ): + return False + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + "Must pass a workspace_id when using a personal_access_token" + ) + + return True diff --git a/seam/utils/token.py b/seam/utils/token.py new file mode 100644 index 00000000..3c94c9f5 --- /dev/null +++ b/seam/utils/token.py @@ -0,0 +1,47 @@ +token_prefix = "seam_" + +access_token_prefix = "seam_at" + +jwt_prefix = "ey" + +client_session_token_prefix = "seam_cst" + +publishable_key_token_prefix = "seam_pk" + + +def is_access_token(token: str) -> bool: + return token.startswith(access_token_prefix) + + +def is_jwt(token: str) -> bool: + return token.startswith(jwt_prefix) + + +def is_seam_token(token: str) -> bool: + return token.startswith(token_prefix) + + +def is_api_key(token: str) -> bool: + return ( + not is_client_session_token(token) + and not is_jwt(token) + and not is_access_token(token) + and not is_publishable_key(token) + and is_seam_token(token) + ) + + +def is_client_session_token(token: str) -> bool: + return token.startswith(client_session_token_prefix) + + +def is_publishable_key(token: str) -> bool: + return token.startswith(publishable_key_token_prefix) + + +def is_console_session_token(token: str) -> bool: + return is_jwt(token) + + +def is_personal_access_token(token: str) -> bool: + return is_access_token(token) From aacf45b835a8b81cfb566a111af4715db66fde1b Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:31:27 +0200 Subject: [PATCH 12/21] Remove unnecessary endpoint cast --- seam/seam.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/seam/seam.py b/seam/seam.py index 44849a34..f235a6fa 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -81,8 +81,6 @@ def __init__( "\033[0m" ) endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT - if endpoint is not None: - self.endpoint = cast(str, endpoint) def make_request(self, method: str, path: str, **kwargs): """ From 962cccd809eb2d47d615d85bf0f1fd87210084c0 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 25 Apr 2024 22:34:06 +0200 Subject: [PATCH 13/21] Update abstract class --- seam/types.py | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/seam/types.py b/seam/types.py index ad374c0f..3a8fec84 100644 --- a/seam/types.py +++ b/seam/types.py @@ -2077,18 +2077,11 @@ def __init__( console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - endpoint: Optional[str] = "https://connect.getseam.com", + endpoint: Optional[str] = None, wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): - self.api_key = api_key - self.client_session_token = client_session_token - self.publishable_key = publishable_key - self.user_identifier_key = user_identifier_key - self.console_session_token = console_session_token - self.personal_access_token = personal_access_token - self.workspace_id = workspace_id - self.endpoint = endpoint self.wait_for_action_attempt = wait_for_action_attempt + self.lts_version = AbstractSeam.lts_version @abc.abstractmethod def make_request(self, method: str, path: str, **kwargs) -> Any: From 8a38158273d26664f518c0a2a339ceafd7dfd1f2 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Mon, 29 Apr 2024 11:41:52 +0200 Subject: [PATCH 14/21] Remove support for publishable key, client session and console session --- seam/seam.py | 74 +---------- seam/types.py | 39 ------ seam/utils/auth.py | 150 +--------------------- seam/utils/options.py | 120 +---------------- seam/utils/parse_options.py | 8 -- test/conftest.py | 2 +- test/test_init_seam.py | 3 - test/workspaces/test_workspaces_create.py | 2 +- 8 files changed, 7 insertions(+), 391 deletions(-) diff --git a/seam/seam.py b/seam/seam.py index f235a6fa..8d475b82 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -3,7 +3,7 @@ from importlib.metadata import version from typing import Optional, Union, Dict, cast from typing_extensions import Self -from seam.utils.auth import get_auth_headers, warn_on_insecure_user_identifier_key +from seam.utils.auth import get_auth_headers from seam.utils.parse_options import get_api_key_from_env, get_endpoint_from_env from .routes import Routes from .types import AbstractSeam, SeamApiException @@ -23,9 +23,6 @@ def __init__( self, api_key: Optional[str] = None, *, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, endpoint: Optional[str] = None, @@ -36,14 +33,6 @@ def __init__( ---------- api_key : str, optional API key. - client_session_token : str, optional - Client session token. - publishable_key : str, optional - Publishable key. - user_identifier_key : str, optional - User identifier key. - console_session_token : str, optional - Console session token. personal_access_token : str, optional Personal access token. workspace_id : str, optional @@ -57,15 +46,10 @@ def __init__( Routes.__init__(self) api_key = api_key or get_api_key_from_env( - client_session_token=client_session_token, - console_session_token=console_session_token, personal_access_token=personal_access_token, ) self.__auth_headers = get_auth_headers( api_key=api_key, - client_session_token=client_session_token, - publishable_key=publishable_key, - console_session_token=console_session_token, personal_access_token=personal_access_token, workspace_id=workspace_id, ) @@ -130,62 +114,6 @@ def from_api_key( api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt ) - @classmethod - def from_client_session_token( - cls, - client_session_token: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - client_session_token=client_session_token, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - - @classmethod - def from_publishable_key( - cls, - publishable_key: str, - user_identifier_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - warn_on_insecure_user_identifier_key(publishable_key) - - seam = cls( - publishable_key=publishable_key, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - client_session = seam.client_sessions.get_or_create( - user_identifier_key=user_identifier_key - ) - - return cls.from_client_session_token( - client_session.token, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - - @classmethod - def from_console_session_token( - cls, - console_session_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - return cls( - console_session_token=console_session_token, - workspace_id=workspace_id, - endpoint=endpoint, - wait_for_action_attempt=wait_for_action_attempt, - ) - @classmethod def from_personal_access_token( cls, diff --git a/seam/types.py b/seam/types.py index 3a8fec84..ad9e5b6f 100644 --- a/seam/types.py +++ b/seam/types.py @@ -2071,10 +2071,6 @@ def __init__( self, api_key: Optional[str] = None, *, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - user_identifier_key: Optional[str] = None, - console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, endpoint: Optional[str] = None, @@ -2098,41 +2094,6 @@ def from_api_key( ) -> Self: raise NotImplementedError - @classmethod - @abc.abstractmethod - def from_client_session_token( - cls, - client_session_token: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_publishable_key( - cls, - publishable_key: str, - user_identifier_key: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - - @classmethod - @abc.abstractmethod - def from_console_session_token( - cls, - console_session_token: str, - workspace_id: str, - *, - endpoint: Optional[str] = None, - wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, - ) -> Self: - raise NotImplementedError - @classmethod @abc.abstractmethod def from_personal_access_token( diff --git a/seam/utils/auth.py b/seam/utils/auth.py index fa6337cb..107049b2 100644 --- a/seam/utils/auth.py +++ b/seam/utils/auth.py @@ -1,12 +1,7 @@ from typing import Optional -import re from seam.utils.options import ( SeamHttpInvalidOptionsError, - is_seam_http_multi_workspace_options_with_personal_access_token, is_seam_http_options_with_api_key, - is_seam_http_options_with_client_session_token, - is_seam_http_multi_workspace_options_with_console_session_token, - is_seam_http_options_with_console_session_token, is_seam_http_options_with_personal_access_token, ) from seam.utils.token import ( @@ -15,10 +10,7 @@ is_client_session_token, is_publishable_key, is_seam_token, - publishable_key_token_prefix, token_prefix, - client_session_token_prefix, - jwt_prefix, access_token_prefix, ) @@ -30,57 +22,18 @@ def __init__(self, message): def get_auth_headers( api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - publishable_key: Optional[str] = None, - console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, ): - if publishable_key: - return get_auth_headers_for_publishable_key(publishable_key) - if is_seam_http_options_with_api_key( api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, personal_access_token=personal_access_token, ): return get_auth_headers_for_api_key(api_key) - if is_seam_http_options_with_client_session_token( - client_session_token=client_session_token, - api_key=api_key, - console_session_token=console_session_token, + if is_seam_http_options_with_personal_access_token( personal_access_token=personal_access_token, - ): - return get_auth_headers_for_client_session_token(client_session_token) - - if is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token=console_session_token, api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - ) or is_seam_http_options_with_console_session_token( - console_session_token=console_session_token, - api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - workspace_id=workspace_id, - ): - return get_auth_headers_for_console_session_token( - console_session_token, workspace_id - ) - - if is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - ) or is_seam_http_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, workspace_id=workspace_id, ): return get_auth_headers_for_personal_access_token( @@ -88,34 +41,12 @@ def get_auth_headers( ) raise SeamHttpInvalidOptionsError( - "Must specify an api_key, client_session_token, publishable_key, console_session_token, " - "or personal_access_token. Attempted reading configuration from the environment, " + "Must specify an api_key or personal_access_token. " + "Attempted reading configuration from the environment, " "but the environment variable SEAM_API_KEY is not set." ) -def get_auth_headers_for_publishable_key(publishable_key: str) -> dict: - if is_jwt(publishable_key): - raise SeamHttpInvalidTokenError("A JWT cannot be used as a publishable_key") - - if is_access_token(publishable_key): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a publishable_key" - ) - - if is_client_session_token(publishable_key): - raise SeamHttpInvalidTokenError( - "A Client Session Token Key cannot be used as a publishable_key" - ) - - if not is_publishable_key(publishable_key): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid publishable_key format, expected token to start with {publishable_key_token_prefix}" - ) - - return {"seam-publishable-key": publishable_key} - - def get_auth_headers_for_api_key(api_key: str) -> dict: if is_client_session_token(api_key): raise SeamHttpInvalidTokenError( @@ -141,63 +72,6 @@ def get_auth_headers_for_api_key(api_key: str) -> dict: return {"authorization": f"Bearer {api_key}"} -def get_auth_headers_for_client_session_token(client_session_token: str) -> dict: - if is_jwt(client_session_token): - raise SeamHttpInvalidTokenError( - "A JWT cannot be used as a client_session_token" - ) - - if is_access_token(client_session_token): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a client_session_token" - ) - - if is_publishable_key(client_session_token): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as a client_session_token" - ) - - if not is_client_session_token(client_session_token): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid client_session_token format, expected token to start with {client_session_token_prefix}" - ) - - return { - "authorization": f"Bearer {client_session_token}", - "client-session-token": client_session_token, - } - - -def get_auth_headers_for_console_session_token( - console_session_token: str, workspace_id: Optional[str] = None -) -> dict: - if is_access_token(console_session_token): - raise SeamHttpInvalidTokenError( - "An Access Token cannot be used as a console_session_token" - ) - - if is_client_session_token(console_session_token): - raise SeamHttpInvalidTokenError( - "A Client Session Token cannot be used as a console_session_token" - ) - - if is_publishable_key(console_session_token): - raise SeamHttpInvalidTokenError( - "A Publishable Key cannot be used as a console_session_token" - ) - - if not is_jwt(console_session_token): - raise SeamHttpInvalidTokenError( - f"Unknown or invalid console_session_token format, expected a JWT which starts with {jwt_prefix}" - ) - - headers = {"authorization": f"Bearer {console_session_token}"} - if workspace_id is not None: - headers["seam-workspace"] = workspace_id - - return headers - - def get_auth_headers_for_personal_access_token( personal_access_token: str, workspace_id: Optional[str] = None ) -> dict: @@ -226,21 +100,3 @@ def get_auth_headers_for_personal_access_token( headers["seam-workspace"] = workspace_id return headers - - -def warn_on_insecure_user_identifier_key(user_identifier_key: str): - if is_email(user_identifier_key): - warning_message = ( - "\033[93m" - "Using an email for the userIdentifierKey is insecure and may return an error in the future!\n" - "This is insecure because an email is common knowledge or easily guessed.\n" - "Use something with sufficient entropy known only to the owner of the client session.\n" - "For help choosing a user identifier key see " - "https://docs.seam.co/latest/seam-components/overview/get-started-with-client-side-components#3-select-a-user-identifier-key" - "\033[0m" - ) - print(warning_message) - - -def is_email(value: str) -> bool: - return re.match(r"^[^\s@]+@[^\s@]+\.[^\s@]+$", value) is not None diff --git a/seam/utils/options.py b/seam/utils/options.py index f51085f8..64bad38c 100644 --- a/seam/utils/options.py +++ b/seam/utils/options.py @@ -8,23 +8,11 @@ def __init__(self, message): def is_seam_http_options_with_api_key( api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, ) -> bool: if api_key is None: return False - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the api_key option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the api_key option" - ) - if personal_access_token is not None: raise SeamHttpInvalidOptionsError( "The personal_access_token option cannot be used with the api_key option" @@ -33,88 +21,10 @@ def is_seam_http_options_with_api_key( return True -def is_seam_http_options_with_client_session_token( - client_session_token: Optional[str] = None, - api_key: Optional[str] = None, - console_session_token: Optional[str] = None, +def is_seam_http_options_with_personal_access_token( personal_access_token: Optional[str] = None, -) -> bool: - if client_session_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the client_session_token option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the client_session_token option" - ) - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the client_session_token option" - ) - - return True - - -def is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token: Optional[str] = None, api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, -) -> bool: - if console_session_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the console_session_token option" - ) - - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the console_session_token option" - ) - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the console_session_token option" - ) - - return True - - -def is_seam_http_options_with_console_session_token( - console_session_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, -) -> bool: - if not is_seam_http_multi_workspace_options_with_console_session_token( - console_session_token=console_session_token, - api_key=api_key, - client_session_token=client_session_token, - personal_access_token=personal_access_token, - ): - return False - - if workspace_id is None: - raise SeamHttpInvalidOptionsError( - "Must pass a workspace_id when using a console_session_token" - ) - - return True - - -def is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, ) -> bool: if personal_access_token is None: return False @@ -124,34 +34,6 @@ def is_seam_http_multi_workspace_options_with_personal_access_token( "The api_key option cannot be used with the personal_access_token option" ) - if client_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The client_session_token option cannot be used with the personal_access_token option" - ) - - if console_session_token is not None: - raise SeamHttpInvalidOptionsError( - "The console_session_token option cannot be used with the personal_access_token option" - ) - - return True - - -def is_seam_http_options_with_personal_access_token( - personal_access_token: Optional[str] = None, - api_key: Optional[str] = None, - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, - workspace_id: Optional[str] = None, -) -> bool: - if not is_seam_http_multi_workspace_options_with_personal_access_token( - personal_access_token=personal_access_token, - api_key=api_key, - client_session_token=client_session_token, - console_session_token=console_session_token, - ): - return False - if workspace_id is None: raise SeamHttpInvalidOptionsError( "Must pass a workspace_id when using a personal_access_token" diff --git a/seam/utils/parse_options.py b/seam/utils/parse_options.py index 4f765e93..19c61ee9 100644 --- a/seam/utils/parse_options.py +++ b/seam/utils/parse_options.py @@ -3,16 +3,8 @@ def get_api_key_from_env( - client_session_token: Optional[str] = None, - console_session_token: Optional[str] = None, personal_access_token: Optional[str] = None, ): - if client_session_token is not None: - return None - - if console_session_token is not None: - return None - if personal_access_token is not None: return None diff --git a/test/conftest.py b/test/conftest.py index 92f2212a..25f981d4 100755 --- a/test/conftest.py +++ b/test/conftest.py @@ -9,6 +9,6 @@ def seam(): r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) seam = Seam( - api_url=f"https://{r}.fakeseamconnect.seam.vc", api_key="seam_apikey1_token" + endpoint=f"https://{r}.fakeseamconnect.seam.vc", api_key="seam_apikey1_token" ) yield seam diff --git a/test/test_init_seam.py b/test/test_init_seam.py index 4fc67b64..7205da38 100644 --- a/test/test_init_seam.py +++ b/test/test_init_seam.py @@ -2,8 +2,5 @@ def test_init_seam_with_fixture(seam: Seam): - assert seam.api_key - assert seam.api_url assert seam.lts_version - assert "http" in seam.api_url assert seam.wait_for_action_attempt is False diff --git a/test/workspaces/test_workspaces_create.py b/test/workspaces/test_workspaces_create.py index a93b2e81..18dd7153 100644 --- a/test/workspaces/test_workspaces_create.py +++ b/test/workspaces/test_workspaces_create.py @@ -6,7 +6,7 @@ def test_workspaces_create(seam: Seam): r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) seam = Seam( - api_url=f"https://{r}.fakeseamconnect.seam.vc", + endpoint=f"https://{r}.fakeseamconnect.seam.vc", api_key="seam_at1_shorttoken_longtoken", ) From df2a67a99bc7956701bacf4212f2bb9c8cfe55b0 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Mon, 29 Apr 2024 17:04:04 +0200 Subject: [PATCH 15/21] Fix tests --- seam/seam.py | 2 +- test/workspaces/test_workspaces_create.py | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/seam/seam.py b/seam/seam.py index 8d475b82..4a590111 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -64,7 +64,7 @@ def __init__( "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." "\033[0m" ) - endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT + self.endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT def make_request(self, method: str, path: str, **kwargs): """ diff --git a/test/workspaces/test_workspaces_create.py b/test/workspaces/test_workspaces_create.py index 18dd7153..a0c5fc55 100644 --- a/test/workspaces/test_workspaces_create.py +++ b/test/workspaces/test_workspaces_create.py @@ -7,7 +7,8 @@ def test_workspaces_create(seam: Seam): r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) seam = Seam( endpoint=f"https://{r}.fakeseamconnect.seam.vc", - api_key="seam_at1_shorttoken_longtoken", + personal_access_token="seam_at1_shorttoken_longtoken", + workspace_id="seed_workspace_1", ) workspace = seam.workspaces.create( From dbb3c5932098d0ed438eab1f5da7c7c5fbd7df6e Mon Sep 17 00:00:00 2001 From: Evan Sosenko Date: Mon, 29 Apr 2024 10:06:11 -0700 Subject: [PATCH 16/21] Update .github/workflows/generate.yml --- .github/workflows/generate.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/generate.yml b/.github/workflows/generate.yml index 780a2612..3b87ef17 100644 --- a/.github/workflows/generate.yml +++ b/.github/workflows/generate.yml @@ -4,7 +4,7 @@ name: Generate on: push: branches-ignore: - - ** + - '**' workflow_dispatch: {} jobs: From b2f6e45c67270ba52405a373ddc94c39e6815843 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Tue, 30 Apr 2024 11:34:45 +0200 Subject: [PATCH 17/21] workspace_id should be required in get_auth_headers_for_personal_access_token --- seam/utils/auth.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/seam/utils/auth.py b/seam/utils/auth.py index 107049b2..fc4f333b 100644 --- a/seam/utils/auth.py +++ b/seam/utils/auth.py @@ -73,7 +73,7 @@ def get_auth_headers_for_api_key(api_key: str) -> dict: def get_auth_headers_for_personal_access_token( - personal_access_token: str, workspace_id: Optional[str] = None + personal_access_token: str, workspace_id: str ) -> dict: if is_jwt(personal_access_token): raise SeamHttpInvalidTokenError( @@ -95,8 +95,7 @@ def get_auth_headers_for_personal_access_token( f"Unknown or invalid personal_access_token format, expected token to start with {access_token_prefix}" ) - headers = {"authorization": f"Bearer {personal_access_token}"} - if workspace_id is not None: - headers["seam-workspace"] = workspace_id - - return headers + return { + "authorization": f"Bearer {personal_access_token}", + "seam-workspace": workspace_id, + } From 4eb625cfa4ae89395c48544bc97d50f95610671e Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Tue, 30 Apr 2024 18:22:54 +0200 Subject: [PATCH 18/21] Add parse_options function --- seam/seam.py | 33 ++++++++++----------------------- seam/utils/parse_options.py | 23 +++++++++++++++++++++++ 2 files changed, 33 insertions(+), 23 deletions(-) diff --git a/seam/seam.py b/seam/seam.py index 4a590111..894ae50c 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -1,17 +1,13 @@ -import os import requests from importlib.metadata import version -from typing import Optional, Union, Dict, cast +from typing import Optional, Union, Dict from typing_extensions import Self -from seam.utils.auth import get_auth_headers -from seam.utils.parse_options import get_api_key_from_env, get_endpoint_from_env + +from seam.utils.parse_options import parse_options from .routes import Routes from .types import AbstractSeam, SeamApiException -DEFAULT_ENDPOINT = "https://connect.getseam.com" - - class Seam(AbstractSeam): """ Initial Seam class used to interact with Seam API @@ -45,26 +41,17 @@ def __init__( Routes.__init__(self) - api_key = api_key or get_api_key_from_env( - personal_access_token=personal_access_token, - ) - self.__auth_headers = get_auth_headers( + self.lts_version = Seam.lts_version + self.wait_for_action_attempt = wait_for_action_attempt + + auth_headers, endpoint = parse_options( api_key=api_key, personal_access_token=personal_access_token, workspace_id=workspace_id, + endpoint=endpoint, ) - self.lts_version = Seam.lts_version - self.wait_for_action_attempt = wait_for_action_attempt - - if os.environ.get("SEAM_API_URL", None) is not None: - print( - "\n" - "\033[93m" - "Using the SEAM_API_URL environment variable is deprecated. " - "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." - "\033[0m" - ) - self.endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT + self.__auth_headers = auth_headers + self.endpoint = endpoint def make_request(self, method: str, path: str, **kwargs): """ diff --git a/seam/utils/parse_options.py b/seam/utils/parse_options.py index 19c61ee9..78d710a5 100644 --- a/seam/utils/parse_options.py +++ b/seam/utils/parse_options.py @@ -1,6 +1,29 @@ import os from typing import Optional +from seam.utils.auth import get_auth_headers + +DEFAULT_ENDPOINT = "https://connect.getseam.com" + + +def parse_options( + api_key: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, + endpoint: Optional[str] = None, +): + api_key = api_key or get_api_key_from_env( + personal_access_token=personal_access_token, + ) + auth_headers = get_auth_headers( + api_key=api_key, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ) + endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT + + return auth_headers, endpoint + def get_api_key_from_env( personal_access_token: Optional[str] = None, From 4d7e3cec1d9a994fda4d171894c042a906b8710f Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 2 May 2024 11:41:54 +0200 Subject: [PATCH 19/21] Move auth and options helpers to seam.py level --- seam/{utils => }/auth.py | 4 +-- seam/{utils => }/parse_options.py | 43 ++++++++++++++++++++++++++++++- seam/seam.py | 2 +- seam/{utils => }/token.py | 0 seam/utils/options.py | 42 ------------------------------ 5 files changed, 45 insertions(+), 46 deletions(-) rename seam/{utils => }/auth.py (97%) rename seam/{utils => }/parse_options.py (56%) rename seam/{utils => }/token.py (100%) delete mode 100644 seam/utils/options.py diff --git a/seam/utils/auth.py b/seam/auth.py similarity index 97% rename from seam/utils/auth.py rename to seam/auth.py index fc4f333b..36bc31f2 100644 --- a/seam/utils/auth.py +++ b/seam/auth.py @@ -1,10 +1,10 @@ from typing import Optional -from seam.utils.options import ( +from seam.parse_options import ( SeamHttpInvalidOptionsError, is_seam_http_options_with_api_key, is_seam_http_options_with_personal_access_token, ) -from seam.utils.token import ( +from seam.token import ( is_jwt, is_access_token, is_client_session_token, diff --git a/seam/utils/parse_options.py b/seam/parse_options.py similarity index 56% rename from seam/utils/parse_options.py rename to seam/parse_options.py index 78d710a5..71f41635 100644 --- a/seam/utils/parse_options.py +++ b/seam/parse_options.py @@ -1,7 +1,7 @@ import os from typing import Optional -from seam.utils.auth import get_auth_headers +from seam.auth import get_auth_headers DEFAULT_ENDPOINT = "https://connect.getseam.com" @@ -55,3 +55,44 @@ def get_endpoint_from_env(): ) return seam_endpoint or seam_api_url + + +class SeamHttpInvalidOptionsError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received invalid options: {message}") + + +def is_seam_http_options_with_api_key( + api_key: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if api_key is None: + return False + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the api_key option" + ) + + return True + + +def is_seam_http_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + workspace_id: Optional[str] = None, +) -> bool: + if personal_access_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + "The api_key option cannot be used with the personal_access_token option" + ) + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + "Must pass a workspace_id when using a personal_access_token" + ) + + return True diff --git a/seam/seam.py b/seam/seam.py index 894ae50c..58d7f664 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -3,7 +3,7 @@ from typing import Optional, Union, Dict from typing_extensions import Self -from seam.utils.parse_options import parse_options +from seam.parse_options import parse_options from .routes import Routes from .types import AbstractSeam, SeamApiException diff --git a/seam/utils/token.py b/seam/token.py similarity index 100% rename from seam/utils/token.py rename to seam/token.py diff --git a/seam/utils/options.py b/seam/utils/options.py deleted file mode 100644 index 64bad38c..00000000 --- a/seam/utils/options.py +++ /dev/null @@ -1,42 +0,0 @@ -from typing import Optional - - -class SeamHttpInvalidOptionsError(Exception): - def __init__(self, message): - super().__init__(f"SeamHttp received invalid options: {message}") - - -def is_seam_http_options_with_api_key( - api_key: Optional[str] = None, - personal_access_token: Optional[str] = None, -) -> bool: - if api_key is None: - return False - - if personal_access_token is not None: - raise SeamHttpInvalidOptionsError( - "The personal_access_token option cannot be used with the api_key option" - ) - - return True - - -def is_seam_http_options_with_personal_access_token( - personal_access_token: Optional[str] = None, - api_key: Optional[str] = None, - workspace_id: Optional[str] = None, -) -> bool: - if personal_access_token is None: - return False - - if api_key is not None: - raise SeamHttpInvalidOptionsError( - "The api_key option cannot be used with the personal_access_token option" - ) - - if workspace_id is None: - raise SeamHttpInvalidOptionsError( - "Must pass a workspace_id when using a personal_access_token" - ) - - return True From 6c817acee3cd380b3fb453188c10bc03179b7636 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 2 May 2024 11:46:18 +0200 Subject: [PATCH 20/21] Simplify parsing pat and api key --- seam/parse_options.py | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/seam/parse_options.py b/seam/parse_options.py index 71f41635..1696646b 100644 --- a/seam/parse_options.py +++ b/seam/parse_options.py @@ -12,9 +12,9 @@ def parse_options( workspace_id: Optional[str] = None, endpoint: Optional[str] = None, ): - api_key = api_key or get_api_key_from_env( - personal_access_token=personal_access_token, - ) + if personal_access_token is None: + api_key = api_key or os.getenv("SEAM_API_KEY") + auth_headers = get_auth_headers( api_key=api_key, personal_access_token=personal_access_token, @@ -25,15 +25,6 @@ def parse_options( return auth_headers, endpoint -def get_api_key_from_env( - personal_access_token: Optional[str] = None, -): - if personal_access_token is not None: - return None - - return os.getenv("SEAM_API_KEY") - - def get_endpoint_from_env(): seam_api_url = os.getenv("SEAM_API_URL") seam_endpoint = os.getenv("SEAM_ENDPOINT") From c9e20ecb2f858002764c534ac210871f6e053022 Mon Sep 17 00:00:00 2001 From: Andrii Balitskyi <10balian10@gmail.com> Date: Thu, 2 May 2024 13:55:56 +0200 Subject: [PATCH 21/21] Export custom errors --- seam/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/seam/__init__.py b/seam/__init__.py index 50537ff1..ff1ef862 100644 --- a/seam/__init__.py +++ b/seam/__init__.py @@ -3,3 +3,5 @@ from seam.seam import Seam from seam.seam import SeamApiException +from seam.parse_options import SeamHttpInvalidOptionsError +from seam.auth import SeamHttpInvalidTokenError