Add support for ST V3

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 All notable changes will be documented in this file.
 
+## [1.5.0] - 2023-10-01
+
+This version adds support for the Service Token V3 (Beta) authentication method for Infisical which is a JSON; note that it continues to support Service Token V2 (the default authentication method at this time). With this update, it's possible to initialize the InfisicalClient with a Service Token V3 JSON via the `tokenJSON` parameter to perform CRUD secret operations.
+
+Example:
+
+```
+client = InfisicalClient(
+    token_json=os.environ.get("INFISICAL_TOKEN_JSON")
+)
+```
+
 ## [1.4.3] - 2023-09-13
 
 This version adds support for the `include_imports` and `attach_to_os_environ` parameters for the `get_all_secrets()` method.

README.md

@@ -84,11 +84,24 @@ from infisical import InfisicalClient
 client = InfisicalClient(token="your_infisical_token")
 ```
 
+Using Infisical Token V3 (Beta):
+
+In `v1.5.0`, we released a superior token authentication method; this credential is a JSON containing a `publicKey`, `privateKey`, and `serviceToken` and can be used to initialize the Node SDK client instead of the regular service token.
+
+You can use this beta feature like so:
+
+```py
+from infisical import InfisicalClient
+
+client = InfisicalClient(token_json="your_infisical_token_v3_json")
+```
+
 ### Options
 
 | Parameter | Type     | Description |
 | --------- | -------- | ----------- |
-| `token`   | `string` | An Infisical Token scoped to a project and environment. |
+| `token`    | `string`  | An Infisical Token scoped to a project and environment(s).                 |
+| `tokenJson`| `string`  | An Infisical Token V3 JSON scoped to a project and environment(s) - in beta|
 | `site_url` | `string` | Your self-hosted Infisical site URL. Default: `https://app.infisical.com`. |
 | `cache_ttl`| `number` | Time-to-live (in seconds) for refreshing cached secrets. Default: `300`.|
 | `debug`   | `boolean` | Turns debug mode on or off. Default: `false`.      |

infisical/api/get_service_token_data_key.py

@@ -0,0 +1,15 @@
+from infisical.api.models import GetServiceTokenKeyResponse
+from requests import Session
+
+
+def get_service_token_data_key_req(
+    api_request: Session,
+) -> GetServiceTokenKeyResponse:
+    """Send request again Infisical API to fetch service token data v3 key.
+
+    :param api_request: The :class:`requests.Session` instance used to perform the request
+    :return: Returns the API response as-is
+    """
+    response = api_request.get("/api/v3/service-token/me/key")
+
+    return GetServiceTokenKeyResponse.parse_obj(response.json())

infisical/api/models.py

@@ -26,3 +26,13 @@ class GetServiceTokenDetailsResponse(BaseModel):
     created_at: datetime = Field(..., alias="createdAt")
     updated_at: datetime = Field(..., alias="updatedAt")
     v: int = Field(..., alias="__v")
+
+class KeyData(BaseModel):
+    id: str = Field(..., alias="_id")
+    workspace: str
+    encrypted_key: str = Field(..., alias="encryptedKey")
+    public_key: str = Field(..., alias="publicKey")
+    nonce: str
+
+class GetServiceTokenKeyResponse(BaseModel):
+    key: KeyData

infisical/client/infisicalclient.py

@@ -1,8 +1,10 @@
+import json
 from typing import Dict, Optional
 
 from infisical.api import create_api_request_with_auth
 from infisical.constants import (
     AUTH_MODE_SERVICE_TOKEN,
+    AUTH_MODE_SERVICE_TOKEN_V3,
     INFISICAL_URL,
     SERVICE_TOKEN_REGEX,
 )
@@ -28,6 +30,7 @@ class InfisicalClient:
     def __init__(
         self,
         token: Optional[str] = None,
+        token_json: Optional[str] = None,
         site_url: str = INFISICAL_URL,
         debug: bool = False,
         cache_ttl: int = 300,
@@ -51,6 +54,20 @@ def __init__(
             )
 
             self.api_request = create_api_request_with_auth(site_url, service_token)
+
+        if token_json and token_json != "":
+            token_dict = json.loads(token_json)
+
+            self.client_config = ClientConfig(
+                auth_mode=AUTH_MODE_SERVICE_TOKEN_V3,
+                credentials={
+                    "public_key": token_dict["publicKey"],
+                    "private_key": token_dict["privateKey"]
+                },
+                cache_ttl=cache_ttl
+            )
+
+            self.api_request = create_api_request_with_auth(site_url, token_dict["serviceToken"])
 
         self.debug = debug
 

infisical/constants.py

@@ -2,5 +2,6 @@
 
 INFISICAL_URL = "https://app.infisical.com"
 AUTH_MODE_SERVICE_TOKEN = "service_token"
+AUTH_MODE_SERVICE_TOKEN_V3 = "service_token_v3"
 
 SERVICE_TOKEN_REGEX = re.compile(r"(st\.[a-f0-9]+\.[a-f0-9]+)\.([a-f0-9]+)")

infisical/models/secret_service.py

@@ -1,17 +1,22 @@
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from pydantic import BaseModel
 from requests import Session
 from typing_extensions import Literal
 
-
 class WorkspaceConfig(BaseModel):
     workspace_id: str
     workspace_key: str
 
+class ServiceTokenCredentials(BaseModel):
+    service_token_key: str
+
+class ServiceTokenV3Credentials(BaseModel):
+    public_key: str
+    private_key: str
 
 class ClientConfig(BaseModel):
-    auth_mode: Literal["service_token"]
-    credentials: Dict[Literal["service_token_key"], str]
+    auth_mode: Literal["service_token", "service_token_v3"]
+    credentials: Union[ServiceTokenCredentials, ServiceTokenV3Credentials]
     workspace_config: Optional[WorkspaceConfig]
     cache_ttl: int

infisical/services/secret_service.py

@@ -7,6 +7,7 @@
 from infisical.api.get_secret import get_secret_req
 from infisical.api.get_secrets import get_secrets_req
 from infisical.api.get_service_token_data import get_service_token_data_req
+from infisical.api.get_service_token_data_key import get_service_token_data_key_req
 from infisical.api.update_secret import update_secret_req
 from infisical.helpers.secrets import transform_secret_to_secret_bundle
 from infisical.models.api import (
@@ -23,6 +24,7 @@
 from infisical.utils.crypto import (
     decrypt_symmetric_128_bit_hex_key_utf8,
     encrypt_symmetric_128_bit_hex_key_utf8,
+    decrypt_asymmetric
 )
 from requests import Session
 from typing_extensions import Literal
@@ -33,19 +35,33 @@ class SecretService:
     def populate_client_config(
         api_request: Session, client_config: ClientConfig
     ) -> WorkspaceConfig:
-        service_token_details = get_service_token_data_req(api_request)
-
-        workspace_key = decrypt_symmetric_128_bit_hex_key_utf8(
-            ciphertext=service_token_details.encrypted_key,
-            iv=service_token_details.iv,
-            tag=service_token_details.tag,
-            key=client_config.credentials["service_token_key"],
-        )
+        if client_config.auth_mode == "service_token":
+            service_token_details = get_service_token_data_req(api_request)
+            workspace_key = decrypt_symmetric_128_bit_hex_key_utf8(
+                ciphertext=service_token_details.encrypted_key,
+                iv=service_token_details.iv,
+                tag=service_token_details.tag,
+                key=client_config.credentials.service_token_key,
+            )
 
-        return WorkspaceConfig(
-            workspace_id=service_token_details.workspace,
-            workspace_key=workspace_key,
-        )
+            return WorkspaceConfig(
+                workspace_id=service_token_details.workspace,
+                workspace_key=workspace_key,
+            )
+
+        if client_config.auth_mode == "service_token_v3":
+            service_token_key_details = get_service_token_data_key_req(api_request)
+            workspace_key = decrypt_asymmetric(
+                ciphertext=service_token_key_details.key.encrypted_key,
+                nonce=service_token_key_details.key.nonce,
+                public_key=service_token_key_details.key.public_key,
+                private_key=client_config.credentials.private_key
+            )
+
+            return WorkspaceConfig(
+                workspace_id=service_token_key_details.key.workspace,
+                workspace_key=workspace_key
+            )
 
     @staticmethod
     def get_fallback_secret(secret_name: str) -> SecretBundle:

tests/test_client/test_infisical_client.py

@@ -7,7 +7,7 @@
 @pytest.fixture(scope="module")
 def client():
     infisical_client = InfisicalClient(
-        token=os.environ["INFISICAL_TOKEN"], site_url=os.environ["SITE_URL"], debug=True
+        token=os.environ.get("INFISICAL_TOKEN"), token_json=os.environ.get("INFISICAL_TOKEN_JSON"), site_url=os.environ.get("SITE_URL"), debug=True
     )
 
     infisical_client.create_secret("KEY_ONE", "KEY_ONE_VAL")