diff --git a/infisical_sdk/__init__.py b/infisical_sdk/__init__.py index 6e7e124..08c9101 100644 --- a/infisical_sdk/__init__.py +++ b/infisical_sdk/__init__.py @@ -1,3 +1,3 @@ from .client import InfisicalSDKClient # noqa from .infisical_requests import InfisicalError # noqa -from .api_types import SingleSecretResponse, ListSecretsResponse, BaseSecret, SymmetricEncryption # noqa \ No newline at end of file +from .api_types import SingleSecretResponse, ListSecretsResponse, BaseSecret, SymmetricEncryption, DynamicSecretProviders # noqa \ No newline at end of file diff --git a/infisical_sdk/api_types.py b/infisical_sdk/api_types.py index 2b0fe29..425a512 100644 --- a/infisical_sdk/api_types.py +++ b/infisical_sdk/api_types.py @@ -293,3 +293,105 @@ def from_dict(cls, data: Dict) -> 'SingleFolderResponse': return cls( folder=SingleFolderResponseItem.from_dict(data['folder']), ) + +class DynamicSecretProviders(str, Enum): + """Enum for dynamic secret provider types""" + AWS_ELASTICACHE = "aws-elasticache" + AWS_IAM = "aws-iam" + AZURE_ENTRA_ID = "azure-entra-id" + AZURE_SQL_DATABASE = "azure-sql-database" + CASSANDRA = "cassandra" + COUCHBASE = "couchbase" + ELASTICSEARCH = "elastic-search" + GCP_IAM = "gcp-iam" + GITHUB = "github" + KUBERNETES = "kubernetes" + LDAP = "ldap" + MONGO_ATLAS = "mongo-db-atlas" + MONGODB = "mongo-db" + RABBITMQ = "rabbit-mq" + REDIS = "redis" + SAP_ASE = "sap-ase" + SAP_HANA = "sap-hana" + SNOWFLAKE = "snowflake" + SQL_DATABASE = "sql-database" + TOTP = "totp" + VERTICA = "vertica" + +@dataclass +class DynamicSecret(BaseModel): + """Infisical Dynamic Secret""" + id: str + name: str + version: int + type: str + folderId: str + createdAt: str + updatedAt: str + defaultTTL: Optional[str] = None + maxTTL: Optional[str] = None + status: Optional[str] = None + statusDetails: Optional[str] = None + usernameTemplate: Optional[str] = None + metadata: Optional[List[Dict[str, str]]] = field(default_factory=list) + inputs: Optional[Any] = None + +@dataclass +class SingleDynamicSecretResponse(BaseModel): + """Response model for get/create/update/delete dynamic secret API""" + dynamicSecret: DynamicSecret + + @classmethod + def from_dict(cls, data: Dict) -> 'SingleDynamicSecretResponse': + return cls( + dynamicSecret=DynamicSecret.from_dict(data['dynamicSecret']), + ) + +@dataclass +class DynamicSecretLease(BaseModel): + """Infisical Dynamic Secret Lease""" + id: str + expireAt: str + createdAt: str + updatedAt: str + version: int + dynamicSecretId: str + externalEntityId: str + status: Optional[str] = None + statusDetails: Optional[str] = None + dynamicSecret: Optional[DynamicSecret] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'DynamicSecretLease': + """Create model from dictionary with nested DynamicSecret""" + lease_data = data.copy() + if 'dynamicSecret' in data and data['dynamicSecret'] is not None: + lease_data['dynamicSecret'] = DynamicSecret.from_dict(data['dynamicSecret']) + + return super().from_dict(lease_data) + +@dataclass +class CreateLeaseResponse(BaseModel): + """Response model for create lease API - returns lease, dynamicSecret, and data""" + lease: DynamicSecretLease + dynamicSecret: DynamicSecret + data: Any + + @classmethod + def from_dict(cls, data: Dict) -> 'CreateLeaseResponse': + return cls( + lease=DynamicSecretLease.from_dict(data['lease']), + dynamicSecret=DynamicSecret.from_dict(data['dynamicSecret']), + data=data.get('data', {}), + ) + +@dataclass +class SingleLeaseResponse(BaseModel): + """Response model for get/delete/renew lease API - returns only lease""" + lease: DynamicSecretLease + + @classmethod + def from_dict(cls, data: Dict) -> 'SingleLeaseResponse': + return cls( + lease=DynamicSecretLease.from_dict(data['lease']), + ) diff --git a/infisical_sdk/client.py b/infisical_sdk/client.py index 1a91c2d..c40da63 100644 --- a/infisical_sdk/client.py +++ b/infisical_sdk/client.py @@ -4,6 +4,7 @@ from infisical_sdk.resources import V3RawSecrets from infisical_sdk.resources import KMS from infisical_sdk.resources import V2Folders +from infisical_sdk.resources import DynamicSecrets from infisical_sdk.util import SecretsCache @@ -26,6 +27,7 @@ def __init__(self, host: str, token: str = None, cache_ttl: int = 60): self.secrets = V3RawSecrets(self.api, self.cache) self.kms = KMS(self.api) self.folders = V2Folders(self.api) + self.dynamic_secrets = DynamicSecrets(self.api) def set_token(self, token: str): """ diff --git a/infisical_sdk/resources/__init__.py b/infisical_sdk/resources/__init__.py index 77a23f5..9f32cb9 100644 --- a/infisical_sdk/resources/__init__.py +++ b/infisical_sdk/resources/__init__.py @@ -1,4 +1,5 @@ from .secrets import V3RawSecrets from .kms import KMS from .auth import Auth -from .folders import V2Folders \ No newline at end of file +from .folders import V2Folders +from .dynamic_secrets import DynamicSecrets \ No newline at end of file diff --git a/infisical_sdk/resources/dynamic_secrets.py b/infisical_sdk/resources/dynamic_secrets.py new file mode 100644 index 0000000..cdd27fc --- /dev/null +++ b/infisical_sdk/resources/dynamic_secrets.py @@ -0,0 +1,335 @@ +from typing import Optional, Dict, Any, List, Union + +from infisical_sdk.infisical_requests import InfisicalRequests +from infisical_sdk.api_types import ( + DynamicSecret, + DynamicSecretLease, + DynamicSecretProviders, + SingleDynamicSecretResponse, + CreateLeaseResponse, + SingleLeaseResponse, +) + + +class DynamicSecretLeases: + """Manages dynamic secret leases.""" + + def __init__(self, requests: InfisicalRequests) -> None: + self.requests = requests + + def create( + self, + dynamic_secret_name: str, + project_slug: str, + environment_slug: str, + path: str = "/", + ttl: str = None) -> CreateLeaseResponse: + """Create a new lease for a dynamic secret. + + Args: + dynamic_secret_name: The name of the dynamic secret to create a lease for. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + ttl: The time to live for the lease (e.g., "1h", "30m"). + + Returns: + CreateLeaseResponse containing lease, dynamicSecret, and data (credentials). + """ + request_body = { + "dynamicSecretName": dynamic_secret_name, + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "ttl": ttl, + } + + result = self.requests.post( + path="/api/v1/dynamic-secrets/leases", + json=request_body, + model=CreateLeaseResponse + ) + + return result.data + + def revoke( + self, + lease_id: str, + project_slug: str, + environment_slug: str, + path: str = "/", + is_forced: bool = False) -> DynamicSecretLease: + """Revoke a dynamic secret lease. + + Args: + lease_id: The ID of the lease to revoke. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + is_forced: A boolean flag to delete the the dynamic secret from Infisical without trying to remove it from external provider. Used when the dynamic secret got modified externally. + + Returns: + The revoked lease. + """ + request_body = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "isForced": is_forced, + } + + result = self.requests.delete( + path=f"/api/v1/dynamic-secrets/leases/{lease_id}", + json=request_body, + model=SingleLeaseResponse + ) + + return result.data.lease + + def renew( + self, + lease_id: str, + project_slug: str, + environment_slug: str, + path: str = "/", + ttl: str = None) -> DynamicSecretLease: + """Renew a dynamic secret lease. + + Args: + lease_id: The ID of the lease to renew. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + ttl: The new time to live for the lease (e.g., "1h", "30m"). + + Returns: + The renewed lease. + """ + request_body = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "ttl": ttl, + } + + result = self.requests.post( + path=f"/api/v1/dynamic-secrets/leases/{lease_id}/renew", + json=request_body, + model=SingleLeaseResponse + ) + + return result.data.lease + + def get( + self, + lease_id: str, + project_slug: str, + environment_slug: str, + path: str = "/") -> DynamicSecretLease: + """Get a dynamic secret lease by ID. + + Args: + lease_id: The ID of the lease to retrieve. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + + Returns: + The lease with dynamicSecret included. + """ + params = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + } + + result = self.requests.get( + path=f"/api/v1/dynamic-secrets/leases/{lease_id}", + params=params, + model=SingleLeaseResponse + ) + + return result.data.lease + + +class DynamicSecrets: + """Manages dynamic secrets in Infisical.""" + + def __init__(self, requests: InfisicalRequests) -> None: + self.requests = requests + self.leases = DynamicSecretLeases(requests) + + def create( + self, + name: str, + provider_type: Union[DynamicSecretProviders, str], + inputs: Dict[str, Any], + default_ttl: str, + max_ttl: str, + project_slug: str, + environment_slug: str, + path: str = "/", + metadata: Optional[List[Dict[str, str]]] = None) -> DynamicSecret: + """Create a new dynamic secret. + + Args: + name: The name of the dynamic secret. + provider_type: The provider type (e.g., DynamicSecretProviders.SQL_DATABASE). + inputs: The provider-specific configuration inputs. Check the Infisical documentation for the specific provider for the inputs: https://infisical.com/docs/api-reference/endpoints/dynamic-secrets/create#body-provider + default_ttl: The default time to live for leases (e.g., "1h", "30m"). + max_ttl: The maximum time to live for leases (e.g., "24h"). + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path where the dynamic secret will be created. Defaults to "/". + metadata: Optional list of metadata items with 'key' and 'value'. + + Returns: + The created dynamic secret. + """ + provider_value = provider_type.value if isinstance(provider_type, DynamicSecretProviders) else provider_type + + request_body = { + "name": name, + "provider": { + "type": provider_value, + "inputs": inputs, + }, + "defaultTTL": default_ttl, + "maxTTL": max_ttl, + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "metadata": metadata, + } + + result = self.requests.post( + path="/api/v1/dynamic-secrets", + json=request_body, + model=SingleDynamicSecretResponse + ) + + return result.data.dynamicSecret + + def delete( + self, + name: str, + project_slug: str, + environment_slug: str, + path: str = "/", + is_forced: bool = False) -> DynamicSecret: + """Delete a dynamic secret. + + Args: + name: The name of the dynamic secret to delete. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + is_forced: A boolean flag to delete the the dynamic secret from Infisical without trying to remove it from external provider. Used when the dynamic secret got modified externally. + + Returns: + The deleted dynamic secret. + """ + request_body = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "isForced": is_forced, + } + + result = self.requests.delete( + path=f"/api/v1/dynamic-secrets/{name}", + json=request_body, + model=SingleDynamicSecretResponse + ) + + return result.data.dynamicSecret + + def get_by_name( + self, + name: str, + project_slug: str, + environment_slug: str, + path: str = "/") -> DynamicSecret: + """Get a dynamic secret by name. + + Args: + name: The name of the dynamic secret. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + + Returns: + The dynamic secret. + """ + params = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + } + + result = self.requests.get( + path=f"/api/v1/dynamic-secrets/{name}", + params=params, + model=SingleDynamicSecretResponse + ) + + return result.data.dynamicSecret + + def update( + self, + name: str, + project_slug: str, + environment_slug: str, + path: str = "/", + default_ttl: Optional[str] = None, + max_ttl: Optional[str] = None, + new_name: Optional[str] = None, + inputs: Optional[Dict[str, Any]] = None, + metadata: Optional[List[Dict[str, str]]] = None, + username_template: Optional[str] = None) -> DynamicSecret: + """Update an existing dynamic secret. + + Args: + name: The current name of the dynamic secret. + project_slug: The slug of the project. + environment_slug: The slug of the environment. + path: The path to the dynamic secret. Defaults to "/". + default_ttl: The new default time to live for leases (e.g., "1h"). + max_ttl: The new maximum time to live for leases (e.g., "24h"). + new_name: The new name for the dynamic secret. + inputs: Updated provider-specific configuration inputs. + metadata: Updated metadata list with 'key' and 'value' items. + username_template: The new username template for the dynamic secret. + + Returns: + The updated dynamic secret. + """ + data: Dict[str, Any] = {} + if inputs is not None: + data["inputs"] = inputs + if default_ttl is not None: + data["defaultTTL"] = default_ttl + if max_ttl is not None: + data["maxTTL"] = max_ttl + if new_name is not None: + data["newName"] = new_name + if metadata is not None: + data["metadata"] = metadata + if username_template is not None: + data["usernameTemplate"] = username_template + + request_body = { + "projectSlug": project_slug, + "environmentSlug": environment_slug, + "path": path, + "data": data, + } + + result = self.requests.patch( + path=f"/api/v1/dynamic-secrets/{name}", + json=request_body, + model=SingleDynamicSecretResponse + ) + + return result.data.dynamicSecret +