Making Use of Vault: Python

import requests

headers = {
    'X-Vault-Request': 'true',
    # change this to your initial root token if you are using my docker vault...
    'X-Vault-Token': 'root',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
}

json_data = {
    'data': {
        'wireguard': {
            'keys': {
                'privkey': 'privkey1',
                'pubkey': 'pubkey2',
            },
            'ip': {
                'outer': '100.64.0.0/30',
                'inner': '100.64.255.0/31',
            },
        },
    },
    'options': {},
}

response = requests.put('http://localhost:8200/v1/secret/data/mycomplicatedsecret', headers=headers, json=json_data)

print(response.json())

import requests

headers = {
    'X-Vault-Request': 'true',
    'X-Vault-Token': 'root',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
}

response = requests.get('http://localhost:8200/v1/secret/data/mycomplicatedsecret', headers=headers)

print(response.json())

>>> from rich import print
>>> print(response.json())
{
    'request_id': 'cf17b7ce-cda7-e8d7-7370-ca757698fd19',
    'lease_id': '',
    'renewable': False,
    'lease_duration': 0,
    'data': {
        'data': {'wireguard': {'ip': {'inner': '100.64.255.0/31', 'outer': '100.64.0.0/30'}, 'keys': {'privkey': 'privkey1', 'pubkey': 'pubkey2'}}},
        'metadata': {'created_time': '2024-11-19T13:50:32.015432Z', 'custom_metadata': None, 'deletion_time': '', 'destroyed': False, 'version': 3}
    },
    'wrap_info': None,
    'warnings': None,
    'auth': None,
    'mount_type': 'kv'
}

from hvac import Client

# where is our vault server?
VAULT_ADDR: str = "http://localhost:8200"
# replace with your initial vault token if you are using docker
VAULT_TOKEN: str = "root"

# lets make an instance of the hvac cault client
vault = Client(url=VAULT_ADDR, token=VAULT_TOKEN)

# lets check if this token is valid and the server is accessible
if not vault.is_authenticated:
    raise RuntimeError("Vault setup failed.")
else:
    print("vault is authenticated!")

# great! lets make a dictionary that represents what we want to update
complex_data: dict = {
    "wireguard": {
        "ip": {"inner": "100.64.255.0/31", "outer": "100.64.0.0/30"},
        "keys": {"privkey": "privkey1", "pubkey": "pubkey2"},
    }
}

# and lets describe where we will be storing this
mount_point = "secret"
secret_path = "mycomplexsecret"

# lets fire the request in that will create_or_update that secret.
results = vault.secrets.kv.v2.create_or_update_secret(
    mount_point=mount_point, path=secret_path, secret=complex_data
)
print(results)

my_version1 = vault.secrets.kv.v2.read_secret_version(mount_point=mount_point, path=secret_path, raise_on_deleted_version=False, version=1)
print(f"v1: {my_version1}")

my_version2 = vault.secrets.kv.v2.read_secret_version(mount_point=mount_point, path=secret_path, raise_on_deleted_version=False, version=1)
print(f"v2: {my_version2}")

from vault_services import (
    quick_vault_auth_from_env,
    export_vault_client_token,
    quick_vault_auth_from_token,
    fetch_kv,
    SecretRequest,
)

# establish a client from env vars
env_client = quick_vault_auth_from_env()

# export its token so I can establish an identical client outside of this 
# execution scope (i.e. across workflow activities)
env_client_token = export_vault_client_token(env_client)

# demonstrate how to launch a client using only the token.
#
# note this is pointless in this contrived example, but if you were using a 
# different auth method like cert or OIDC, you might not enjoy doing this 
# over and again, where using an token from some other authenticated session
# is easier by far.
token_client = quick_vault_auth_from_token(token=env_client_token)

# build a SecretRequest using our pydantic model
my_request_on_env_client = SecretRequest(
    vault_token=env_client_token,
    secret_mount="secret",
    secret_path="mysimplesecret",
    secret_name="username",
)

# send the request and get the response back (which is also a pydantic model)
env_client_response = fetch_kv(secret_request=my_request_on_env_client)

# print it out for you to see it.
print(
    f"you asked for \"{my_request_on_env_client.model_dump()['secret_name']}\" and the result was \"{env_client_response.model_dump()['value']}\""
)

from hvac import Client
from os import environ
from pydantic import BaseModel

# constants that _could_ come from env
# vault token
if not environ.get("VAULT_TOKEN"):
    VAULT_TOKEN: str = "root"
else:
    VAULT_TOKEN: str = environ.get("VAULT_TOKEN")
# vault address
if not environ.get("VAULT_ADDR"):
    VAULT_ADDR = "http://127.0.0.1:8200"
else:
    VAULT_ADDR: str = environ.get("VAULT_ADDR")
# vault tls true/false
if not environ.get("VAULT_INSECURE"):
    VAULT_INSECURE: True
else:
    VAULT_INSECURE: bool = environ.get("VAULT_INSECURE")

# constants that are in the script only
VAULT_TOKEN_MIN_TTL = 30


# pydantic model that describes a secret request
class SecretRequest(BaseModel):
    vault_token: str
    secret_mount: str
    secret_path: str
    secret_name: str


# pydantic model that describes a secret response
class SecretResponse(BaseModel):
    key: str
    value: str


def quick_vault_auth_from_env() -> Client:
    """
    Leverage an env var and login to vault quickly

    Raises:
        RuntimeError: Vault is sad.

    Returns:
        hvac.Client: Authenticated Vault client
    """
    # spin an instance of the client
    vault_client = Client(url=VAULT_ADDR, token=VAULT_TOKEN, verify=not VAULT_INSECURE)
    # bypass for dev token
    if VAULT_TOKEN == "root":
        pass
    else:
        # check if the token is valid and alive
        if not _check_vault_token_ttl_ok(vault_client):
            # we have less than 30s - elaborate.
            if not _check_vault_token_expired(vault_client):
                # we are not expired, but we have less than 30s - renew
                _renew_vault_token(vault_client)
            else:
                raise Exception("Token is already expired!")
        else:
            # we have > 30s of validity - carry on.
            pass
    # if we got here, we have a valid token with > 30s of life; done.
    return vault_client


def quick_vault_auth_from_token(token: str) -> Client:
    """
    Leverage a provided token and login to vault quickly

    Raises:
        RuntimeError: Vault is sad.

    Returns:
        hvac.Client: Authenticated Vault client
    """
    # spin an instance of the client
    vault_client = Client(url=VAULT_ADDR, token=token, verify=not VAULT_INSECURE)
    # bypass for dev token
    if token == "root":
        pass
    else:
        # check if the token is valid and alive
        if not _check_vault_token_ttl_ok(vault_client):
            # we have less than 30s - elaborate.
            if not _check_vault_token_expired(vault_client):
                # we are not expired, but we have less than 30s - renew
                _renew_vault_token(vault_client)
            else:
                raise Exception("Token is already expired!")
        else:
            # we have > 30s of validity - carry on.
            pass
    # if we got here, we have a valid token with > 30s of life; done.
    return vault_client


def export_vault_client_token(vault_client: Client) -> str:
    """
    take the vault client and return just the token as a string

    Returns:
        str: Vault Token in string format
    """
    return vault_client.token


def _check_vault_token_ttl_ok(vault_client: Client) -> bool:
    """
    test a vault client object is actually authenticated to the server
     and has at least 30 seconds of TTL left
     (as defined by constant VAULT_MIN_TTL)

    Returns:
        bool: if the Vault Client object is
    """
    if vault_client.is_authenticated():
        current_token = vault_client.auth.token.lookup_self()
        return current_token["data"]["ttl"] > VAULT_TOKEN_MIN_TTL
    else:
        return False


def _check_vault_token_expired(vault_client: Client) -> bool:
    """
    test a vault client object is actually authenticated to the server
     and has

    Returns:
        bool: if the Vault Client object is
    """
    if vault_client.is_authenticated():
        current_token = vault_client.auth.token.lookup_self()
        return current_token["data"]["ttl"] > 0
    else:
        return False


def _renew_vault_token(vault_client: Client) -> bool:
    """
    renew (restart the ttl clock) on a vault client's token.

    Returns:
        bool: if renew was successful
    """
    if vault_client.is_authenticated():
        vault_client.auth.token.renew_self()
        return True
    else:
        return False


def fetch_kv(secret_request: SecretRequest) -> SecretResponse:
    """
    take a secret request and fetch the named secret.

    Args:
        secret_request (SecretRequest): parameters for the request to vault

    Raises:
        Exception: error relating to the failure to get the secret

    Returns:
        SecretResponse: typed object with the key/value response
    """
    vault_client = quick_vault_auth_from_token(
        token=secret_request.model_dump()["vault_token"]
    )
    # get our secret block from the store
    secrets = vault_client.secrets.kv.v2.read_secret_version(
        mount_point=secret_request.model_dump()["secret_mount"],
        path=secret_request.model_dump()["secret_path"],
    )
    # check if our named secret is there or not
    if secrets["data"]["data"].get(secret_request.model_dump()["secret_name"]) is None:
        raise Exception(
            f"Failed to retrieve token {secret_request.model_dump()['secret_name']} from mount point {secret_request.model_dump()['secret_mount']} and path {secret_request.model_dump()['secret_path']}"
        )

    return SecretResponse(
        key=secret_request.model_dump()["secret_name"],
        value=secrets["data"]["data"][secret_request.model_dump()["secret_name"]],
    )