read invoker credentials from Vault

import os

import re

import requests

class VaultError(Exception):

    """Vault could not be reached or returned an unexpected response"""

# invoker_id comes from the URL and becomes part of the Vault path. Restricting it to

# [A-Za-z0-9_-] prevents path traversal (e.g. "..%2f..%2fca") out of the user's subtree.

# Do not relax this without another guard.

_INVOKER_ID_RE = re.compile(r"[A-Za-z0-9_-]+")

def vault_addr():

    """Base URL of Vault. VAULT_ADDR, otherwise falls back to http://VAULT_HOSTNAME:VAULT_PORT as before."""

    addr = os.getenv("VAULT_ADDR")

        return False

    return True

def read_invoker(user_uuid, invoker_id):

    """Read an invoker's stored credentials from Vault KV v2.

    Returns the stored dict (private_key, certificate, api_invoker_id), or None if

    no such invoker exists for this user. Raises VaultError on any other failure,

    so callers can tell "not found" (404) apart from "Vault is down" (5xx).

    """

    if not _INVOKER_ID_RE.fullmatch(invoker_id or ""):

        return None

    path = f"secret/data/backoffice/users/{user_uuid}/invokers/{invoker_id}"

    try:

        response = requests.get(

            f"{vault_addr()}/v1/{path}",

            headers=vault_headers(),

            verify=vault_verify(),

            timeout=10,

        )

    except requests.exceptions.RequestException as e:

        raise VaultError(f"Vault read failed for invoker {invoker_id}: {e}") from e

    if response.status_code == 404:

        return None

    if not response.ok:

        raise VaultError(f"Vault read failed for invoker {invoker_id}: {response.status_code}")

    try:

        return response.json()["data"]["data"]

    except (ValueError, KeyError, TypeError) as e:

        raise VaultError(f"Unexpected Vault response shape for invoker {invoker_id}") from e

 No newline at end of file