vaultRecursiveSecretsFetcher.py recursively fetches all secrets on a given path and prints them on the terminal in a tabulated format

vaultRecursiveSecretsFetcher.py

import hvac
import os
from tabulate import tabulate
from hvac.exceptions import VaultError

# Initialize the HashiCorp Vault client using environment variables for address and token
client = hvac.Client(
    url=os.environ['VAULT_ADDR'],  # Vault server address
    token=os.environ['VAULT_TOKEN']  # Authentication token
)

def read_secret(client, path, mount_path):
    """Reads a secret from a specified path and mount point."""
    try:
        # Attempt to read the secret version at the specified path
        read_response = client.secrets.kv.read_secret_version(
            path=path,
            mount_point=mount_path
        )
        # Return the data portion of the secret
        return read_response['data']['data']
    except VaultError as e:
        # Handle errors (e.g., permission issues, nonexistent paths) by printing an error message
        print(f"Error reading secret at path: {path} with mount path: {mount_path}. Error: {e}")
        return {}

def list_secrets(client, path=None, mount_path=None):
    """Lists all secrets at a specified path and mount point."""
    try:
        # Attempt to list secrets at the specified path
        list_response = client.secrets.kv.v2.list_secrets(
            path=path,
            mount_point=mount_path
        )
        return list_response
    except VaultError as e:
        # Handle errors by printing an error message and returning None
        print(f"Error listing secrets at path: {path} with mount path: {mount_path}. Error: {e}")
        return None

def list_all_secrets(client, secrets_details, path=None, mount_path=None):
    """Recursively lists all secrets and their details, accumulating results in secrets_details."""
    response = list_secrets(client, path=path, mount_path=mount_path)
    if response is None:
        return

    keys = response['data']['keys']

    for key in keys:
        _path = f"{path}{key}" if path else key  # Construct the full path for the current key

        if key.endswith("/"):
            # If the key is a directory, recurse into it
            list_all_secrets(client, secrets_details, path=_path, mount_path=mount_path)
        else:
            # If the key is a secret, read its value and store its details
            secrets = read_secret(client, path=_path, mount_path=mount_path)
            vault_ui_url = f"{os.environ['VAULT_ADDR']}/ui/vault/secrets/{mount_path.strip('/')}/show/{_path.strip('/')}"
            secrets_details.append({"path": _path, "secrets": secrets, "url": vault_ui_url})

def print_secrets_table(secrets_details):
    """Prints a formatted table of secrets, including paths, keys, values, and Vault UI URLs."""
    rows = []
    for detail in secrets_details:
        for key, value in detail["secrets"].items():
            rows.append([detail["path"], key, str(value), detail["url"]])
    headers = ["Secret Path", "Key", "Value", "Vault UI URL"]
    print(tabulate(rows, headers=headers, tablefmt="grid"))

# List to accumulate secret details
secrets_details = []

# List all secrets starting from the root path in the 'kv' mount, and print them
list_all_secrets(client, secrets_details, path='/', mount_path='kv/')
print_secrets_table(secrets_details)