From 1f47da7dd301239cf94da77fa59750a3fe408e29 Mon Sep 17 00:00:00 2001 From: Scott Wilson Date: Wed, 5 Feb 2025 17:41:17 -0800 Subject: [PATCH 1/2] feature: KMS operation support + docs --- .gitignore | 3 + README.md | 146 ++++++++++++++++++++++++++++++ example.py | 2 +- infisical_sdk/api_types.py | 63 ++++++++++++- infisical_sdk/client.py | 178 ++++++++++++++++++++++++++++++++++++- 5 files changed, 388 insertions(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index 43995bd..cb79de0 100644 --- a/.gitignore +++ b/.gitignore @@ -64,3 +64,6 @@ target/ #Ipython Notebook .ipynb_checkpoints + +# IDEs +.idea diff --git a/README.md b/README.md index 3927eea..3beb895 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ +from infisical_sdk.api_types import SymmetricEncryptionfrom infisical_sdk.api_types import KmsKeysOrderBy + # Infisical Python SDK The Infisical SDK provides a convenient way to interact with the Infisical API. @@ -208,3 +210,147 @@ deleted_secret = client.secrets.delete_secret_by_name( **Returns:** - `BaseSecret`: The response after deleting the secret. + +### `kms` + +This sub-class handles KMS related operations: + +#### List KMS Keys + +```python +kms_keys = client.kms.list_keys( + project_id="", + offset=0, # Optional + limit=100, # Optional + order_by=KmsKeysOrderBy.NAME, # Optional + order_direction=OrderDirection.ASC, # Optional + search=None # Optional +) +``` + +**Parameters:** +- `project_id` (str): The ID of your project. +- `offset` (int, optional): The offset to paginate from. +- `limit` (int, optional): The page size for paginating. +- `order_by` (KmsKeysOrderBy, optional): The key property to order the list response by. +- `order_direction` (OrderDirection, optional): The direction to order the list response in. +- `search` (str, optional): The text value to filter key names by. + +**Returns:** +- `ListKmsKeysResponse`: The response containing the list of KMS keys. + +#### Get KMS Key by ID + +```python +kms_key = client.kms.get_key_by_id( + key_id="" +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to retrieve. + +**Returns:** +- `KmsKey`: The specified key. + +#### Get KMS Key by Name + +```python +kms_key = client.kms.get_key_by_name( + key_name="my-key", + project_id="" +) +``` + +**Parameters:** +- `key_name` (str): The name of the key to retrieve. +- `project_id` (str): The ID of your project. + +**Returns:** +- `KmsKey`: The specified key. + +#### Create KMS Key + +```python +kms_key = client.kms.create_key( + name="my-key", + project_id="", + encryption_algorithm=SymmetricEncryption.AES_GCM_256, + description=None # Optional +) +``` + +**Parameters:** +- `name` (str): The name of the key (must be slug-friendly). +- `project_id` (str): The ID of your project. +- `encryption_algorithm` (SymmetricEncryption): The encryption alogrithm this key should use. +- `description` (str, optional): A description of your key. + +**Returns:** +- `KmsKey`: The newly created key. + +#### Update KMS Key + +```python +updated_key = client.kms.update_key( + key_id="", + name="my-updated-key", # Optional + description="Updated description", # Optional + is_disabled=True # Optional +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to be updated. +- `name` (str, optional): The updated name of the key (must be slug-friendly). +- `description` (str): The updated description of the key. +- `is_disabled` (str): The flag to disable operations with this key. + +**Returns:** +- `KmsKey`: The updated key. + +#### Delete KMS Key + +```python +deleted_key = client.kms.delete_key( + key_id="" +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to be deleted. + +**Returns:** +- `KmsKey`: The deleted key. + +#### Encrypt Data with KMS Key + +```python +encrypted_data = client.kms.encrypt_data( + key_id="", + plaintext="TXkgc2VjcmV0IG1lc3NhZ2U=" # must be base64 encoded +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to encrypt the data with. +- `plaintext` (str): The plaintext data to encrypt (must be base64 encoded). + +**Returns:** +- `str`: The encrypted ciphertext. + +#### Decrypte Data with KMS Key + +```python +decrypted_data = client.kms.decrypt_data( + key_id="", + ciphertext="Aq96Ry7sMH3k/ogaIB5MiSfH+LblQRBu69lcJe0GfIvI48ZvbWY+9JulyoQYdjAx" +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to decrypt the data with. +- `ciphertext` (str): The ciphertext returned from the encrypt operation. + +**Returns:** +- `str`: The base64 encoded plaintext. \ No newline at end of file diff --git a/example.py b/example.py index 61054b5..ac820b8 100644 --- a/example.py +++ b/example.py @@ -2,7 +2,7 @@ sdkInstance = InfisicalSDKClient(host="https://app.infisical.com") -sdkInstance.auth.universalAuth.login("<>", "<>") +sdkInstance.auth.universal_auth.login("<>", "<>") # new_secret = sdkInstance.secrets.create_secret_by_name( # secret_name="NEW_SECRET", diff --git a/infisical_sdk/api_types.py b/infisical_sdk/api_types.py index d1eec39..50c8617 100644 --- a/infisical_sdk/api_types.py +++ b/infisical_sdk/api_types.py @@ -112,7 +112,7 @@ class SingleSecretResponse(BaseModel): secret: BaseSecret @classmethod - def from_dict(cls, data: Dict) -> 'ListSecretsResponse': + def from_dict(cls, data: Dict) -> 'SingleSecretResponse': return cls( secret=BaseSecret.from_dict(data['secret']), ) @@ -125,3 +125,64 @@ class MachineIdentityLoginResponse(BaseModel): expiresIn: int accessTokenMaxTTL: int tokenType: str + +class SymmetricEncryption(str, Enum): + AES_GCM_256 = "aes-256-gcm" + AES_GCM_128 = "aes-128-gcm" + +class OrderDirection(str, Enum): + ASC = "asc" + DESC = "desc" + +class KmsKeysOrderBy(str, Enum): + NAME = "name" + +@dataclass +class KmsKey(BaseModel): + """Infisical KMS Key""" + id: str + description: str + isDisabled: bool + orgId: str + name: str + createdAt: str + updatedAt: str + projectId: str + version: int + encryptionAlgorithm: SymmetricEncryption + +@dataclass +class ListKmsKeysResponse(BaseModel): + """Complete response model for Kms Keys API""" + keys: List[KmsKey] + totalCount: int + + @classmethod + def from_dict(cls, data: Dict) -> 'ListKmsKeysResponse': + """Create model from dictionary with camelCase keys, handling nested objects""" + return cls( + keys=[KmsKey.from_dict(key) for key in data['keys']], + totalCount=data['totalCount'] + ) + + +@dataclass +class SingleKmsKeyResponse(BaseModel): + """Response model for get/create/update/delete API""" + key: KmsKey + + @classmethod + def from_dict(cls, data: Dict) -> 'SingleKmsKeyResponse': + return cls( + key=KmsKey.from_dict(data['key']), + ) + +@dataclass +class KmsKeyEncryptDataResponse(BaseModel): + """Response model for encrypt data API""" + ciphertext: str + +@dataclass +class KmsKeyDecryptDataResponse(BaseModel): + """Response model for decrypt data API""" + plaintext: str \ No newline at end of file diff --git a/infisical_sdk/client.py b/infisical_sdk/client.py index 97cc2db..5a1637f 100644 --- a/infisical_sdk/client.py +++ b/infisical_sdk/client.py @@ -12,7 +12,8 @@ from botocore.exceptions import NoCredentialsError from .infisical_requests import InfisicalRequests -from .api_types import ListSecretsResponse, MachineIdentityLoginResponse +from .api_types import ListSecretsResponse, MachineIdentityLoginResponse, ListKmsKeysResponse, SingleKmsKeyResponse, \ + KmsKey, SymmetricEncryption, KmsKeyEncryptDataResponse, KmsKeyDecryptDataResponse, KmsKeysOrderBy, OrderDirection from .api_types import SingleSecretResponse, BaseSecret @@ -25,6 +26,7 @@ def __init__(self, host: str, token: str = None): self.auth = Auth(self) self.secrets = V3RawSecrets(self) + self.kms = KMS(self) def set_token(self, token: str): """ @@ -307,7 +309,7 @@ def update_secret_by_name( "secretPath": secret_path, "secretValue": secret_value, "secretComment": secret_comment, - "new_secret_name": new_secret_name, + "newSecretName": new_secret_name, "tagIds": None, "skipMultilineEncoding": skip_multiline_encoding, "type": "shared", @@ -343,3 +345,175 @@ def delete_secret_by_name( ) return result.data.secret + +class KMS: + def __init__(self, client: InfisicalSDKClient) -> None: + self.client = client + + def list_keys( + self, + project_id: str, + offset: int = 0, + limit: int = 100, + order_by: KmsKeysOrderBy = KmsKeysOrderBy.NAME, + order_direction: OrderDirection = OrderDirection.ASC, + search: str = None) -> ListKmsKeysResponse: + + params = { + "projectId": project_id, + "search": search, + "offset": offset, + "limit": limit, + "orderBy": order_by, + "orderDirection": order_direction, + } + + result = self.client.api.get( + path="/api/v1/kms/keys", + params=params, + model=ListKmsKeysResponse + ) + + return result.data + + def get_key_by_id( + self, + key_id: str) -> KmsKey: + + result = self.client.api.get( + path=f"/api/v1/kms/keys/{key_id}", + model=SingleKmsKeyResponse + ) + + return result.data.key + + def get_key_by_name( + self, + key_name: str, + project_id: str) -> KmsKey: + + params = { + "projectId": project_id, + } + + result = self.client.api.get( + path=f"/api/v1/kms/keys/key-name/{key_name}", + params=params, + model=SingleKmsKeyResponse + ) + + return result.data.key + + def create_key( + self, + name: str, + project_id: str, + encryption_algorithm: SymmetricEncryption, + description: str = None) -> KmsKey: + + request_body = { + "name": name, + "projectId": project_id, + "encryptionAlgorithm": encryption_algorithm, + "description": description, + } + + result = self.client.api.post( + path="/api/v1/kms/keys", + json=request_body, + model=SingleKmsKeyResponse + ) + + return result.data.key + + def update_key( + self, + key_id: str, + name: str = None, + is_disabled: bool = None, + description: str = None) -> KmsKey: + + request_body = { + "name": name, + "isDisabled": is_disabled, + "description": description, + } + + result = self.client.api.patch( + path=f"/api/v1/kms/keys/{key_id}", + json=request_body, + model=SingleKmsKeyResponse + ) + + return result.data.key + + def delete_key( + self, + key_id: str) -> KmsKey: + + result = self.client.api.delete( + path=f"/api/v1/kms/keys/{key_id}", + json={}, + model=SingleKmsKeyResponse + ) + + return result.data.key + + def encrypt_data( + self, + key_id: str, + plaintext: str) -> str: + """ + Encrypt data with the specified KMS key. + + :param key_id: The ID of the key to decrypt the ciphertext with + :type key_id: str + :param plaintext: The base64 encoded plaintext to encrypt + :type plaintext: str + + + :return: The encrypted base64 encoded plaintext (ciphertext) + :rtype: str + """ + + request_body = { + "plaintext": plaintext + } + + result = self.client.api.post( + path=f"/api/v1/kms/keys/{key_id}/encrypt", + json=request_body, + model=KmsKeyEncryptDataResponse + ) + + return result.data.ciphertext + + def decrypt_data( + self, + key_id: str, + ciphertext: str) -> str: + """ + Decrypt data with the specified KMS key. + + :param key_id: The ID of the key to decrypt the ciphertext with + :type key_id: str + :param ciphertext: The encrypted base64 plaintext to decrypt + :type ciphertext: str + + + :return: The base64 encoded plaintext + :rtype: str + """ + + request_body = { + "ciphertext": ciphertext + } + + + result = self.client.api.post( + path=f"/api/v1/kms/keys/{key_id}/decrypt", + json=request_body, + model=KmsKeyDecryptDataResponse + ) + + return result.data.plaintext \ No newline at end of file From a1901a03c384d4b1d801666da649b92a221b3c86 Mon Sep 17 00:00:00 2001 From: Daniel Hougaard Date: Thu, 6 Feb 2025 06:23:55 +0400 Subject: [PATCH 2/2] fix: minor improvements --- README.md | 6 ++---- infisical_sdk/api_types.py | 9 ++++++++- infisical_sdk/client.py | 18 ++++++++++-------- 3 files changed, 20 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 3beb895..2b6f847 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,3 @@ -from infisical_sdk.api_types import SymmetricEncryptionfrom infisical_sdk.api_types import KmsKeysOrderBy - # Infisical Python SDK The Infisical SDK provides a convenient way to interact with the Infisical API. @@ -328,13 +326,13 @@ deleted_key = client.kms.delete_key( ```python encrypted_data = client.kms.encrypt_data( key_id="", - plaintext="TXkgc2VjcmV0IG1lc3NhZ2U=" # must be base64 encoded + base64EncodedPlaintext="TXkgc2VjcmV0IG1lc3NhZ2U=" # must be base64 encoded ) ``` **Parameters:** - `key_id` (str): The ID of the key to encrypt the data with. -- `plaintext` (str): The plaintext data to encrypt (must be base64 encoded). +- `base64EncodedPlaintext` (str): The plaintext data to encrypt (must be base64 encoded). **Returns:** - `str`: The encrypted ciphertext. diff --git a/infisical_sdk/api_types.py b/infisical_sdk/api_types.py index 50c8617..de5bd4e 100644 --- a/infisical_sdk/api_types.py +++ b/infisical_sdk/api_types.py @@ -126,17 +126,21 @@ class MachineIdentityLoginResponse(BaseModel): accessTokenMaxTTL: int tokenType: str + class SymmetricEncryption(str, Enum): AES_GCM_256 = "aes-256-gcm" AES_GCM_128 = "aes-128-gcm" + class OrderDirection(str, Enum): ASC = "asc" DESC = "desc" + class KmsKeysOrderBy(str, Enum): NAME = "name" + @dataclass class KmsKey(BaseModel): """Infisical KMS Key""" @@ -151,6 +155,7 @@ class KmsKey(BaseModel): version: int encryptionAlgorithm: SymmetricEncryption + @dataclass class ListKmsKeysResponse(BaseModel): """Complete response model for Kms Keys API""" @@ -177,12 +182,14 @@ def from_dict(cls, data: Dict) -> 'SingleKmsKeyResponse': key=KmsKey.from_dict(data['key']), ) + @dataclass class KmsKeyEncryptDataResponse(BaseModel): """Response model for encrypt data API""" ciphertext: str + @dataclass class KmsKeyDecryptDataResponse(BaseModel): """Response model for decrypt data API""" - plaintext: str \ No newline at end of file + plaintext: str diff --git a/infisical_sdk/client.py b/infisical_sdk/client.py index 5a1637f..e830a5f 100644 --- a/infisical_sdk/client.py +++ b/infisical_sdk/client.py @@ -12,9 +12,11 @@ from botocore.exceptions import NoCredentialsError from .infisical_requests import InfisicalRequests -from .api_types import ListSecretsResponse, MachineIdentityLoginResponse, ListKmsKeysResponse, SingleKmsKeyResponse, \ - KmsKey, SymmetricEncryption, KmsKeyEncryptDataResponse, KmsKeyDecryptDataResponse, KmsKeysOrderBy, OrderDirection -from .api_types import SingleSecretResponse, BaseSecret + +from .api_types import ListSecretsResponse, SingleSecretResponse, BaseSecret +from .api_types import SymmetricEncryption, KmsKeysOrderBy, OrderDirection +from .api_types import ListKmsKeysResponse, SingleKmsKeyResponse, MachineIdentityLoginResponse +from .api_types import KmsKey, KmsKeyEncryptDataResponse, KmsKeyDecryptDataResponse class InfisicalSDKClient: @@ -346,6 +348,7 @@ def delete_secret_by_name( return result.data.secret + class KMS: def __init__(self, client: InfisicalSDKClient) -> None: self.client = client @@ -462,13 +465,13 @@ def delete_key( def encrypt_data( self, key_id: str, - plaintext: str) -> str: + base64EncodedPlaintext: str) -> str: """ Encrypt data with the specified KMS key. :param key_id: The ID of the key to decrypt the ciphertext with :type key_id: str - :param plaintext: The base64 encoded plaintext to encrypt + :param base64EncodedPlaintext: The base64 encoded plaintext to encrypt :type plaintext: str @@ -477,7 +480,7 @@ def encrypt_data( """ request_body = { - "plaintext": plaintext + "plaintext": base64EncodedPlaintext } result = self.client.api.post( @@ -509,11 +512,10 @@ def decrypt_data( "ciphertext": ciphertext } - result = self.client.api.post( path=f"/api/v1/kms/keys/{key_id}/decrypt", json=request_body, model=KmsKeyDecryptDataResponse ) - return result.data.plaintext \ No newline at end of file + return result.data.plaintext