vendor_connectors.vault
Vault Connector using jbcom ecosystem packages.
Submodules
Section titled “Submodules”Package Contents
Section titled “Package Contents”Classes
Section titled “Classes”VaultConnector | Vault connector with token and AppRole authentication. |
|---|
VAULT_URL_ENV_VAR | |
|---|---|
VAULT_NAMESPACE_ENV_VAR | |
VAULT_ROLE_ID_ENV_VAR | |
VAULT_SECRET_ID_ENV_VAR | |
VAULT_APPROLE_PATH_ENV_VAR |
vendor_connectors.vault.VAULT_URL_ENV_VAR
Section titled “vendor_connectors.vault.VAULT_URL_ENV_VAR”‘VAULT_ADDR’
vendor_connectors.vault.VAULT_NAMESPACE_ENV_VAR
Section titled “vendor_connectors.vault.VAULT_NAMESPACE_ENV_VAR”‘VAULT_NAMESPACE’
vendor_connectors.vault.VAULT_ROLE_ID_ENV_VAR
Section titled “vendor_connectors.vault.VAULT_ROLE_ID_ENV_VAR”‘VAULT_ROLE_ID’
vendor_connectors.vault.VAULT_SECRET_ID_ENV_VAR
Section titled “vendor_connectors.vault.VAULT_SECRET_ID_ENV_VAR”‘VAULT_SECRET_ID’
vendor_connectors.vault.VAULT_APPROLE_PATH_ENV_VAR
Section titled “vendor_connectors.vault.VAULT_APPROLE_PATH_ENV_VAR”‘VAULT_APPROLE_PATH’
class vendor_connectors.vault.VaultConnector(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, logger: lifecyclelogging.Logging | None = None, **kwargs)
Section titled “class vendor_connectors.vault.VaultConnector(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, logger: lifecyclelogging.Logging | None = None, **kwargs)”Bases: vendor_connectors.base.VendorConnectorBase
Vault connector with token and AppRole authentication.
Initialization
Section titled “Initialization”Initialize the connector.
Args: api_key: API key (overrides environment variable) base_url: Base URL (overrides class default) timeout: HTTP timeout in seconds logger: Logger instance **kwargs: Passed to DirectedInputsClass
property vault_client : hvac.Client
Section titled “property vault_client : hvac.Client”Lazy initialization of the Vault client.
_set_token_expiration()
Section titled “_set_token_expiration()”Set the token expiration time.
_is_token_valid() → bool
Section titled “_is_token_valid() → bool”Check if the current Vault token is still valid.
static _validate_mount_point(mount_point: str | None) → None
Section titled “static _validate_mount_point(mount_point: str | None) → None”Ensure Vault mount inputs do not allow traversal or null bytes.
classmethod get_vault_client(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, **kwargs) → hvac.Client
Section titled “classmethod get_vault_client(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, **kwargs) → hvac.Client”Get an instance of the Vault client.
list_secrets(root_path: str = ’/’, mount_point: str = ‘secret’, max_depth: int | None = None) → dict[str, dict]
Section titled “list_secrets(root_path: str = ’/’, mount_point: str = ‘secret’, max_depth: int | None = None) → dict[str, dict]”List secrets recursively from Vault KV v2 engine.
Args: root_path: Starting path for listing (default: “/”). mount_point: KV engine mount point (default: “secret”). max_depth: Maximum directory depth to traverse (None = unlimited).
Returns: Dict mapping secret paths to their data.
Raises: ValueError: If root_path contains path traversal sequences.
read_secret(path: str, mount_point: str = ‘secret’) → dict | None
Section titled “read_secret(path: str, mount_point: str = ‘secret’) → dict | None”Read a single secret from Vault.
Args: path: Path to the secret. mount_point: KV engine mount point (default: “secret”).
Returns: Secret data dict, or None if not found.
get_secret(path: str = ’/’, secret_name: str | None = None, matchers: dict[str, str] | None = None, mount_point: str = ‘secret’) → dict | None
Section titled “get_secret(path: str = ’/’, secret_name: str | None = None, matchers: dict[str, str] | None = None, mount_point: str = ‘secret’) → dict | None”Get Vault secret by path, name, or by searching with matchers.
This method supports three modes:
- Direct path + secret_name: Fetches secret at path/secret_name
- Path with matchers: Searches secrets under path and returns first match
- Path without matchers: Returns first non-empty secret found
Args: path: Root path to search or base path for secret_name (default: “/”). secret_name: Specific secret name to append to path. matchers: Dict of key/value pairs to match against secret data. mount_point: KV engine mount point (default: “secret”).
Returns: Secret data dict, or None if not found.
write_secret(path: str, data: dict, mount_point: str = ‘secret’) → bool
Section titled “write_secret(path: str, data: dict, mount_point: str = ‘secret’) → bool”Write a secret to Vault.
Args: path: Path to write the secret. data: Secret data dict. mount_point: KV engine mount point (default: “secret”).
Returns: True if successful, False otherwise.
list_aws_iam_roles(mount_point: str = ‘aws’, name_prefix: str | None = None) → list[str]
Section titled “list_aws_iam_roles(mount_point: str = ‘aws’, name_prefix: str | None = None) → list[str]”List AWS IAM roles configured in Vault’s AWS secrets engine.
Args: mount_point: AWS secrets engine mount point (default: “aws”). name_prefix: Optional prefix filter for role names.
Returns: List of role names available for credential generation.
get_aws_iam_role(role_name: str, mount_point: str = ‘aws’) → dict | None
Section titled “get_aws_iam_role(role_name: str, mount_point: str = ‘aws’) → dict | None”Retrieve details about a specific AWS IAM role configured in Vault.
Args: role_name: Name of the role to fetch. mount_point: AWS secrets engine mount point (default: “aws”).
Returns: Dict containing the role configuration, or None if not found.
generate_aws_credentials(role_name: str, mount_point: str = ‘aws’, ttl: str | None = None, credential_type: str | None = None) → dict[str, Any]
Section titled “generate_aws_credentials(role_name: str, mount_point: str = ‘aws’, ttl: str | None = None, credential_type: str | None = None) → dict[str, Any]”Generate AWS credentials via Vault’s AWS secrets engine.
Args: role_name: AWS role configured in Vault. mount_point: AWS secrets engine mount point (default: “aws”). ttl: Optional TTL override (e.g., “1h”). credential_type: Optional credential type override (e.g., “sts”).
Returns: Dict of generated credential data (e.g., AccessKeyId, SecretAccessKey, SessionToken).
Raises: ValueError: If role_name is empty or mount_point is invalid. RuntimeError: If Vault fails to return credentials.