tPj VdZddlmZddlmZddlZddlmZddlm Z Gdde Z y) ag Custom Secret Manager Integration This module provides a base class for implementing custom secret managers in LiteLLM. Usage: from litellm.integrations.custom_secret_manager import CustomSecretManager class MySecretManager(CustomSecretManager): def __init__(self): super().__init__(secret_manager_name="my_secret_manager") async def async_read_secret( self, secret_name: str, optional_params=None, timeout=None, ): # Your implementation here return await self._fetch_secret_from_service(secret_name) def sync_read_secret( self, secret_name: str, optional_params=None, timeout=None, ): # Your implementation here return self._fetch_secret_from_service_sync(secret_name) # Set your custom secret manager import litellm from litellm.types.secret_managers.main import KeyManagementSystem litellm.secret_manager_client = MySecretManager() litellm._key_management_system = KeyManagementSystem.CUSTOM )abstractmethod)AnyN)verbose_logger)BaseSecretManagerceZdZdZ ddedzffd Ze ddededzdee jzdzdedzfd Z e ddededzdee jzdzdedzfd Z dded ed edzdedzdee jzdzd ee zdzdeeeffdZ ddededzdedzdee jzdzdef dZdefdZddee jzdzdefdZdefdZxZS)CustomSecretManageray Base class for implementing custom secret managers. This class provides a standard interface for implementing custom secret management integrations in LiteLLM. Users can extend this class to integrate their own secret management systems. Example: ```python from litellm.integrations.custom_secret_manager import CustomSecretManager class MyVaultSecretManager(CustomSecretManager): def __init__(self, vault_url: str, token: str): super().__init__(secret_manager_name="my_vault") self.vault_url = vault_url self.token = token async def async_read_secret(self, secret_name: str, optional_params=None, timeout=None): # Implementation for reading secrets from your vault async with httpx.AsyncClient() as client: response = await client.get( f"{self.vault_url}/v1/secret/{secret_name}", headers={"X-Vault-Token": self.token}, timeout=timeout ) return response.json()["data"]["value"] def sync_read_secret(self, secret_name: str, optional_params=None, timeout=None): # Sync implementation with httpx.Client() as client: response = client.get( f"{self.vault_url}/v1/secret/{secret_name}", headers={"X-Vault-Token": self.token}, timeout=timeout ) return response.json()["data"]["value"] ``` Nsecret_manager_namec bt||xsd|_tjdy)a5 Initialize the CustomSecretManager. Args: secret_manager_name: A descriptive name for your secret manager. This is used for logging and debugging purposes. **kwargs: Additional keyword arguments to pass to your secret manager. custom_secret_managerz!Initialized custom secret managerN)super__init__r rinfo)selfr kwargs __class__s o/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/integrations/custom_secret_manager.pyr zCustomSecretManager.__init__Xs- #6#Q:Q ?@ secret_nameoptional_paramstimeoutreturnc Kyw)a Asynchronously read a secret from your custom secret manager. Args: secret_name: Name/path of the secret to read optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: The secret value if found, None otherwise Raises: Exception: If there's an error reading the secret Nrrrrs rasync_read_secretz%CustomSecretManager.async_read_secretisscy)a Synchronously read a secret from your custom secret manager. Args: secret_name: Name/path of the secret to read optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: The secret value if found, None otherwise Raises: Exception: If there's an error reading the secret Nrrs rsync_read_secretz$CustomSecretManager.sync_read_secretsr secret_value descriptiontagsc<Ktd|jdw)a Asynchronously write a secret to your custom secret manager. This is optional to implement. If your secret manager supports writing secrets, you can override this method. Args: secret_name: Name/path of the secret to write secret_value: Value to store description: Description of the secret optional_params: Additional parameters specific to your secret manager timeout: Request timeout tags: Optional tags to apply to the secret Returns: Response from the secret manager containing write operation details Raises: NotImplementedError: If write operations are not supported z)Write operations are not implemented for z5. Override async_write_secret() to add write support.NotImplementedErrorr )rrrrrrr s rasync_write_secretz&CustomSecretManager.async_write_secrets0:"78P8P7QRB B  recovery_window_in_daysc<Ktd|jdw)a Asynchronously delete a secret from your custom secret manager. This is optional to implement. If your secret manager supports deleting secrets, you can override this method. Args: secret_name: Name of the secret to delete recovery_window_in_days: Number of days before permanent deletion (if supported) optional_params: Additional parameters specific to your secret manager timeout: Request timeout Returns: Response from the secret manager containing deletion details Raises: NotImplementedError: If delete operations are not supported z*Delete operations are not implemented for z7. Override async_delete_secret() to add delete support.r")rrr&rrs rasync_delete_secretz'CustomSecretManager.async_delete_secrets02"89Q9Q8RSD D  r%c.tjdy)a= Validate that all required environment variables and configuration are present. Override this method to validate your secret manager's configuration. Returns: True if the environment is valid Raises: ValueError: If required configuration is missing z>No environment validation configured for custom secret managerT)rdebugrs rvalidate_environmentz(CustomSecretManager.validate_environments ]^rcLKtjd|jyw)a# Perform a health check on your secret manager. This is optional to implement. Override this method to add health check support. Args: timeout: Request timeout Returns: True if the secret manager is healthy, False otherwise z#Health check not implemented for %sT)rr*r )rrs rasync_health_checkz&CustomSecretManager.async_health_checks" BDD\D\]s"$cPd|jjd|jdS)N)r__name__r r+s r__repr__zCustomSecretManager.__repr__s)4>>**+6$2J2J1K2NNr)N)NN)NNNN)NN)r1 __module__ __qualname____doc__strr rdictfloathttpxTimeoutrrlistrr$intr(boolr,r.r2 __classcell__)rs@rrr0s%R+/A 4ZA"(,04    &-  t   *(,04    &-  t   2#''+04#'      4Z    &-  TkD    c3h  J/0'+04   "%t   &-    < d   0E0L X\ O#Orr) r6abcrtypingrr:litellm._loggingr+litellm.secret_managers.base_secret_managerrrrrrrDs-$L +IDO+DOr