Vault MCP Bridge

from typing import Optional, Dict, Any, List
import hvac
from .settings import settings
from .models import Principal

def new_vault_client() -> hvac.Client:
    client = hvac.Client(url=settings.VAULT_ADDR, namespace=settings.VAULT_NAMESPACE or None)
    if settings.VAULT_TOKEN:
        client.token = settings.VAULT_TOKEN
    elif settings.VAULT_ROLE_ID and settings.VAULT_SECRET_ID:
        resp = client.auth.approle.login(role_id=settings.VAULT_ROLE_ID, secret_id=settings.VAULT_SECRET_ID)
        client.token = resp["auth"]["client_token"]
    else:
        raise RuntimeError("No Vault auth configured (token or AppRole required)")
    if not client.is_authenticated():
        raise RuntimeError("Vault auth failed")
    return client

def client_for_principal(principal: Principal) -> hvac.Client:
    parent = new_vault_client()
    if not settings.CHILD_TOKEN_ENABLED:
        return parent
    policy_name = f"{settings.CHILD_TOKEN_POLICY_PREFIX}{principal.subject}"
    res = parent.auth.token.create(policies=[policy_name], ttl=settings.CHILD_TOKEN_TTL, display_name=f"mcp-{principal.subject}")
    token = (
        (res.get("auth") or {}).get("client_token")
        or (res.get("auth") or {}).get("token")
        or (res.get("data") or {}).get("token")
        or res.get("client_token")
    )
    if not token:
        return parent
    child = hvac.Client(url=settings.VAULT_ADDR, namespace=settings.VAULT_NAMESPACE or None)
    child.token = token
    return child

def kv_read(client: hvac.Client, path: str, version: Optional[int] = None) -> Dict[str, Any]:
    return client.secrets.kv.v2.read_secret_version(mount_point=settings.KV_MOUNT, path=path, version=version)

def kv_write(client: hvac.Client, path: str, data: Dict[str, Any], cas: Optional[int] = None):
    return client.secrets.kv.v2.create_or_update_secret(mount_point=settings.KV_MOUNT, path=path, secret=data, cas=cas)

def kv_delete_latest(client: hvac.Client, path: str):
    return client.secrets.kv.v2.delete_latest_version_of_secret(mount_point=settings.KV_MOUNT, path=path)

def kv_list(client: hvac.Client, path: str) -> List[str]:
    res = client.secrets.kv.v2.list_secrets(mount_point=settings.KV_MOUNT, path=path)
    return res.get("data", {}).get("keys", [])

def kv_undelete(client: hvac.Client, path: str, versions: List[int]):
    return client.secrets.kv.v2.undelete_secret_versions(mount_point=settings.KV_MOUNT, path=path, versions=versions)

def kv_destroy(client: hvac.Client, path: str, versions: List[int]):
    return client.secrets.kv.v2.destroy_secret_versions(mount_point=settings.KV_MOUNT, path=path, versions=versions)