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..2b6f847 100644 --- a/README.md +++ b/README.md @@ -208,3 +208,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="", + base64EncodedPlaintext="TXkgc2VjcmV0IG1lc3NhZ2U=" # must be base64 encoded +) +``` + +**Parameters:** +- `key_id` (str): The ID of the key to encrypt the data with. +- `base64EncodedPlaintext` (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..de5bd4e 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,71 @@ 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 diff --git a/infisical_sdk/client.py b/infisical_sdk/client.py index 97cc2db..e830a5f 100644 --- a/infisical_sdk/client.py +++ b/infisical_sdk/client.py @@ -12,8 +12,11 @@ from botocore.exceptions import NoCredentialsError from .infisical_requests import InfisicalRequests -from .api_types import ListSecretsResponse, MachineIdentityLoginResponse -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: @@ -25,6 +28,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 +311,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 +347,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, + 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 base64EncodedPlaintext: The base64 encoded plaintext to encrypt + :type plaintext: str + + + :return: The encrypted base64 encoded plaintext (ciphertext) + :rtype: str + """ + + request_body = { + "plaintext": base64EncodedPlaintext + } + + 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