hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

kms: migrate hanzo-kms off Infisical paths onto the luxfi/kms surface

Browse Source 
kms.hanzo.ai runs luxfi/kms, which has never served an /api/* route. This
SDK was sending Infisical's: /api/v3/secrets/raw for every secret operation,
/api/v3/auth/login for user auth, /api/v1/auth/kubernetes-auth/login for k8s.
Only one call in the whole package — /v1/kms/auth/login — was ever real.

It stayed invisible because older luxfi/kms builds embedded a console SPA
behind a root catch-all that answered every unmatched path with 200
text/html. A wrong URL came back as a JSON decode error, which reads like a
parsing bug in the client rather than "this endpoint does not exist." That
catch-all is gone (cmd/kms/main.go ends in notFoundJSON), so those paths now
return honest JSON 404s and the SDK fails outright.

The route table is not the only thing that was wrong. The data model was
Infisical's too — project_id, workspace, secret version, secret comment,
shared-vs-personal type — and luxfi/kms has none of those concepts. It keys
one value by (org, path, name, env) in ZapDB at kms/secrets/{path}/{env}/{name}
and a write upserts it in place. Half-migrating would have left response
models that cannot validate what the server sends, so the vocabulary moves
with the paths:

  list_secrets(path, env) -> names   GET  /v1/kms/orgs/{org}/secrets
  get_secret(path, name, env)        GET  /v1/kms/orgs/{org}/secrets/{path}/{name}
  put_secret(path, name, value, env) POST /v1/kms/orgs/{org}/secrets   (create AND replace)
  delete_secret(path, name, env)     DELETE .../secrets/{path}/{name}
  health()                           GET  /v1/kms/healthz

Two server behaviors now have one home, hanzo_kms/routes.py, shared by the
sync and async clients so they stay mirror images:

  - The server splits the trailing path at its LAST slash into (path, name),
    so each segment is escaped individually. Escaping the joined string
    encodes the separators away and the server reads one long name. A name
    containing "/" is rejected outright: it would be written under one key
    and read back under another, so the write looks like it succeeded and
    the read never finds it.
  - There is no versioned read. get_secret(version=N) raises
    VersionUnsupportedError rather than quietly returning the current value.

org is new and required — it scopes both the URL and the JWT owner claim.
Constructor field, HANZO_KMS_ORG, defaults to "hanzo".

Auth collapses to what the server actually offers: client credentials
exchanged at /v1/kms/auth/login, or a pre-issued IAM bearer token. The AWS,
Azure, GCP, Kubernetes and SRP methods were Infisical's and none were served.
That also fixes a divergence where the async client silently ignored
HANZO_KMS_TOKEN — env parsing is now one function.

Callers migrated with it: hanzo-cli's `hanzo kms`, the hanzo-tools-kms MCP
tool, and hanzo-tools-auth's session client, all of which were building the
removed auth models and calling the removed create/update pair. Each now
constructs KMSClient() and lets it read the environment.

Tests: pkg/hanzo-kms/tests pins the wire shape — no request URL may contain
"/api/", per-segment escaping survives the server's last-slash split, sync
and async emit byte-identical request lines, and a decoder rejects the
200-HTML/JSON-404 shapes instead of reading them as empty results. Wired into
hanzo-packages-ci without `|| true`, so a regression turns CI red instead of
waiting for a decode error in production.

hanzo-kms 1.1.0 -> 1.1.1.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-07-26 08:29:59 -07:00
co-authored by Hanzo Dev
parent c53d6a7c81
commit c80140810a
15 changed files with 1224 additions and 1492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/hanzo-packages-ci.yml
+32 -1
View File
@@ -13,6 +13,7 @@ on:
- 'pkg/hanzo-aci/**'
- 'pkg/hanzo-dev/**'
- 'pkg/hanzoai/**'
- 'pkg/hanzo-kms/**'
- '.github/workflows/hanzo-packages-ci.yml'
tags:
- 'v*'
@@ -35,6 +36,7 @@ on:
- 'pkg/hanzo-aci/**'
- 'pkg/hanzo-dev/**'
- 'pkg/hanzoai/**'
- 'pkg/hanzo-kms/**'
- '.github/workflows/hanzo-packages-ci.yml'
jobs:
@@ -149,7 +151,36 @@ jobs:
- name: Run ALL tests
run: |
timeout 120 python -m pytest tests/ -v --tb=short --cov=hanzo_memory --cov-report=term-missing -n 4 --maxfail=5 -k "not TestInfinityClient" || true # Parallel with 2min timeout
test-hanzo-kms:
name: Test hanzo-kms
runs-on: hanzo-build-linux-amd64
timeout-minutes: 5
defaults:
run:
working-directory: pkg/hanzo-kms
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e .[dev]
# No `|| true`: these tests pin the canonical luxfi/kms path shape. A
# regression back onto Infisical's /api/* surface has to turn CI red —
# in production it was invisible, because the console SPA behind the
# server's old root catch-all answered every wrong path with 200 HTML.
- name: Run tests
run: |
timeout 120 python -m pytest tests/ -v --tb=short
integration-test:
name: Integration Test
needs: [test-hanzo-network, test-hanzo-mcp, test-hanzo-aci, test-hanzo-memory, test-hanzo, test-hanzo-dev]
LLM.md
+8
View File
@@ -30,6 +30,14 @@ uv run pytest tests/ -v # tests
- `pkg/hanzo-mcp/` — MCP server; tools via `[project.entry-points."hanzo.tools"]`.
- `pkg/hanzo-tools-*/` — one concern each, exports a `TOOLS` list.
- `pkg/hanzo-{agents,agent,network,memory}/` — agent/compute/memory libraries.
- `pkg/hanzo-kms/` — KMS client. The server is **luxfi/kms** (`kms.hanzo.ai`,
`kms.lux.network`) and its whole surface is `/v1/kms/auth/login` plus
`/v1/kms/orgs/{org}/secrets[/{path}/{name}]`. `/api/*` is Infisical's and was
never served — it looked like a decode error rather than a 404 only because old
builds answered every unmatched path with the console SPA (200 text/html).
A secret is (org, path, name, env), one value each — no versions. The server
splits the trailing URL at its LAST slash into (path, name), so escape each
segment individually. `pkg/hanzo-kms/tests/` pins all of it.
**Rules for agents:** update THIS file with significant discoveries; never write
random summary files; keep the README cross-link block intact.
pkg/hanzo-cli/hanzo_cli/kms/commands.py
+78 -150
View File
@@ -1,11 +1,14 @@
"""Hanzo CLI — KMS subcommands for secret management.
A secret is (org, path, name, env). The org comes from HANZO_KMS_ORG
(default: hanzo); env defaults to "default".
Usage:
hanzo kms list <project> <env> — List secrets
hanzo kms get <project> <env> <name> — Get secret value
hanzo kms set <project> <env> <name> <value> — Create or update secret
hanzo kms delete <project> <env> <name> — Delete a secret
hanzo kms inject <project> <env> — Print export statements
hanzo kms list --path providers/lux --env prod — List secret names
hanzo kms get providers/lux deploy-mnemonic — Get a secret value
hanzo kms set providers/lux deploy-mnemonic VALUE — Create or replace
hanzo kms delete providers/lux deploy-mnemonic — Delete a secret
hanzo kms inject --path providers/lux — Print export statements
"""
from __future__ import annotations
@@ -16,224 +19,149 @@ from rich.table import Table
console = Console()
DEFAULT_ENV = "default"
def _get_kms_client():
"""Build a KMS client from environment or stored auth."""
import os
"""Build a KMS client from the environment.
from hanzo_kms import ClientSettings, KMSClient
KMSClient() reads HANZO_KMS_URL / _ORG / _CLIENT_ID / _CLIENT_SECRET /
_TOKEN itself — see hanzo_kms.settings_from_env.
"""
from hanzo_kms import KMSClient
# KMS has its own auth (Universal Auth with client_id/secret)
# Check for dedicated KMS env vars first
kms_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
if client_id and client_secret:
from hanzo_kms import AuthenticationOptions, UniversalAuthMethod
settings = ClientSettings(
site_url=kms_url,
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id=client_id,
client_secret=client_secret,
)
),
)
return KMSClient(settings=settings)
# Fall back to default env-based construction
return KMSClient()
def _masked(value: str) -> str:
return f"{value[:4]}***" if len(value) > 4 else "***"
@click.group()
def kms() -> None:
"""Manage secrets via Hanzo KMS."""
@kms.command("list")
@click.argument("project")
@click.argument("env")
@click.option("--path", default="/", help="Secret path prefix.")
def list_secrets(project: str, env: str, path: str) -> None:
"""List all secrets in a project environment."""
@click.option("--path", default="", help="Secret path, e.g. providers/lux.")
@click.option("--env", default=DEFAULT_ENV, help="Environment bucket.")
def list_secrets(path: str, env: str) -> None:
"""List the secret names at a path.
Names only — the server's list route returns names, not values.
"""
client = _get_kms_client()
try:
secrets = client.list_secrets(
project_id=project,
environment=env,
path=path,
)
names = client.list_secrets(path, env)
finally:
client.close()
if not secrets:
if not names:
console.print("[yellow]No secrets found.[/yellow]")
return
table = Table(title=f"Secrets: {project} / {env} ({path})")
table.add_column("Key", style="cyan")
table.add_column("Value", style="dim")
table.add_column("Version", style="white", justify="right")
table.add_column("Updated", style="white")
for s in secrets:
# Mask the value — show first 4 chars then ***
val = s.secret_value
masked = f"{val[:4]}***" if len(val) > 4 else "***"
updated = str(s.updated_at)[:19] if s.updated_at else "—"
table.add_row(s.secret_key, masked, str(s.version), updated)
table = Table(title=f"Secrets: {path or '/'} ({env})")
table.add_column("Name", style="cyan")
for name in names:
table.add_row(name)
console.print(table)
@kms.command("get")
@click.argument("project")
@click.argument("env")
@click.argument("path")
@click.argument("name")
@click.option("--path", default="/", help="Secret path prefix.")
@click.option("--env", default=DEFAULT_ENV, help="Environment bucket.")
@click.option("--reveal", is_flag=True, help="Show full value (default: masked).")
def get_secret(project: str, env: str, name: str, path: str, reveal: bool) -> None:
def get_secret(path: str, name: str, env: str, reveal: bool) -> None:
"""Get a single secret's value."""
client = _get_kms_client()
try:
secret = client.get_secret(
project_id=project,
environment=env,
secret_name=name,
path=path,
)
value = client.get_secret(path, name, env)
finally:
client.close()
if reveal:
console.print(secret.secret_value)
else:
val = secret.secret_value
masked = f"{val[:4]}***" if len(val) > 4 else "***"
table = Table(title=f"{name}")
table.add_column("Field", style="cyan")
table.add_column("Value", style="white")
table.add_row("Key", secret.secret_key)
table.add_row("Value", masked)
table.add_row("Version", str(secret.version))
table.add_row("Type", secret.type)
table.add_row("Environment", secret.environment)
if secret.secret_comment:
table.add_row("Comment", secret.secret_comment)
console.print(table)
console.print(value if reveal else _masked(value))
@kms.command("set")
@click.argument("project")
@click.argument("env")
@click.argument("path")
@click.argument("name")
@click.argument("value", required=False)
@click.option("--path", default="/", help="Secret path prefix.")
@click.option("--comment", default=None, help="Secret comment.")
def set_secret(
project: str,
env: str,
name: str,
value: str | None,
path: str,
comment: str | None,
) -> None:
"""Create or update a secret. Prompts for value if not given."""
@click.option("--env", default=DEFAULT_ENV, help="Environment bucket.")
def set_secret(path: str, name: str, value: str | None, env: str) -> None:
"""Create or replace a secret. Prompts for the value if not given.
One upsert — KMS holds exactly one value per (path, name, env).
"""
if not value:
value = click.prompt("Secret value", hide_input=True)
client = _get_kms_client()
try:
# Try update first, create if it doesn't exist
try:
secret = client.update_secret(
project_id=project,
environment=env,
secret_name=name,
secret_value=value,
)
console.print(f"[green]Updated[/green] {name} (v{secret.version})")
except Exception:
kwargs = {}
if comment:
kwargs["secret_comment"] = comment
secret = client.create_secret(
project_id=project,
environment=env,
secret_name=name,
secret_value=value,
**kwargs,
)
console.print(f"[green]Created[/green] {name}")
client.put_secret(path, name, value, env)
finally:
client.close()
console.print(f"[green]Set[/green] {path}/{name} ({env})")
@kms.command("delete")
@click.argument("project")
@click.argument("env")
@click.argument("path")
@click.argument("name")
@click.option("--path", default="/", help="Secret path prefix.")
@click.option("--env", default=DEFAULT_ENV, help="Environment bucket.")
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation.")
def delete_secret(project: str, env: str, name: str, path: str, yes: bool) -> None:
def delete_secret(path: str, name: str, env: str, yes: bool) -> None:
"""Delete a secret."""
if not yes:
click.confirm(f"Delete secret '{name}' from {project}/{env}?", abort=True)
click.confirm(f"Delete secret '{path}/{name}' from {env}?", abort=True)
client = _get_kms_client()
try:
client.delete_secret(
project_id=project,
environment=env,
secret_name=name,
path=path,
)
client.delete_secret(path, name, env)
finally:
client.close()
console.print(f"[green]Deleted[/green] {name}")
console.print(f"[green]Deleted[/green] {path}/{name} ({env})")
@kms.command("inject")
@click.argument("project")
@click.argument("env")
@click.option("--path", default="/", help="Secret path prefix.")
@click.option("--path", default="", help="Secret path.")
@click.option("--env", default=DEFAULT_ENV, help="Environment bucket.")
@click.option(
"--format", "fmt", type=click.Choice(["export", "dotenv", "json"]), default="export"
)
def inject_secrets(project: str, env: str, path: str, fmt: str) -> None:
def inject_secrets(path: str, env: str, fmt: str) -> None:
"""Print secrets as export statements, dotenv, or JSON.
Pipe to `eval` or redirect to .env file:
hanzo kms inject myproject production | source /dev/stdin
hanzo kms inject myproject production --format dotenv > .env
Pipe to `eval` or redirect to a .env file:
hanzo kms inject --path providers/lux | source /dev/stdin
hanzo kms inject --path providers/lux --format dotenv > .env
Costs one list request plus one read per secret — the server's list
route returns names, not values.
"""
import json
client = _get_kms_client()
try:
secrets = client.list_secrets(
project_id=project,
environment=env,
path=path,
)
values = {
name: client.get_secret(path, name, env)
for name in client.list_secrets(path, env)
}
finally:
client.close()
if not secrets:
if not values:
return
if fmt == "export":
for s in secrets:
# Shell-safe quoting
val = s.secret_value.replace("'", "'\\''")
click.echo(f"export {s.secret_key}='{val}'")
elif fmt == "dotenv":
for s in secrets:
val = s.secret_value.replace('"', '\\"')
click.echo(f'{s.secret_key}="{val}"')
elif fmt == "json":
data = {s.secret_key: s.secret_value for s in secrets}
click.echo(json.dumps(data, indent=2))
if fmt == "json":
click.echo(json.dumps(values, indent=2))
return
for name, value in values.items():
if fmt == "export":
shell_safe = value.replace("'", "'\\''")
click.echo(f"export {name}='{shell_safe}'")
else: # dotenv
dotenv_safe = value.replace('"', '\\"')
click.echo(f'{name}="{dotenv_safe}"')
pkg/hanzo-kms/README.md
+85 -85
View File
@@ -1,6 +1,9 @@
# Hanzo KMS - Python SDK
Official Python SDK for [Hanzo KMS](https://kms.hanzo.ai) - Secret management for your applications.
Official Python SDK for [Hanzo KMS](https://kms.hanzo.ai) — secret management for your applications.
The server is [luxfi/kms](https://github.com/luxfi/kms); this SDK speaks its
canonical `/v1/kms` surface and nothing else.
## Installation
@@ -16,108 +19,87 @@ uv add hanzo-kms
## Quick Start
A secret is identified by **(org, path, name, env)**. The org scopes both the
URL and the JWT; path, name and env identify the value.
```python
from hanzo_kms import KMSClient, ClientSettings, AuthenticationOptions, UniversalAuthMethod
from hanzo_kms import ClientSettings, KMSClient
# Initialize client
client = KMSClient(ClientSettings(
site_url="https://kms.hanzo.ai",
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id="your-client-id",
client_secret="your-client-secret",
)
)
org="lux",
client_id="your-client-id",
client_secret="your-client-secret",
))
# List all secrets
secrets = client.list_secrets(
project_id="my-project",
environment="production"
)
# List the secret names at a path
names = client.list_secrets("providers/lux", env="prod")
for secret in secrets:
print(f"{secret.secret_key}: {secret.secret_value}")
# Read one value
mnemonic = client.get_secret("providers/lux", "deploy-mnemonic", env="prod")
# Get a specific secret
db_url = client.get_value(
project_id="my-project",
environment="production",
secret_name="DATABASE_URL"
)
# Create or replace — one upsert; KMS holds one value per (path, name, env)
client.put_secret("providers/lux", "deploy-mnemonic", mnemonic, env="prod")
# Inject all secrets into environment
client.inject_env(
project_id="my-project",
environment="production"
)
# Delete
client.delete_secret("providers/lux", "deploy-mnemonic", env="prod")
# Load a whole path into os.environ, keyed by secret name
client.inject_env("providers/lux", env="prod")
```
`env` defaults to `"default"`.
## Async
`AsyncKMSClient` is the mirror image of `KMSClient` — same methods, same
arguments.
```python
from hanzo_kms import AsyncKMSClient
async with AsyncKMSClient() as client:
names = await client.list_secrets("providers/lux", env="prod")
```
## Environment Variables
The client can be configured via environment variables:
```bash
export HANZO_KMS_URL="https://kms.hanzo.ai"
export HANZO_KMS_URL="https://kms.hanzo.ai" # default
export HANZO_KMS_ORG="hanzo" # default
export HANZO_KMS_CLIENT_ID="your-client-id"
export HANZO_KMS_CLIENT_SECRET="your-client-secret"
# ...or a pre-issued IAM bearer token instead of client credentials:
export HANZO_KMS_TOKEN="eyJ..."
```
Then simply:
Then:
```python
from hanzo_kms import KMSClient
client = KMSClient() # Uses environment variables
secrets = client.list_secrets("my-project", "production")
client = KMSClient() # configures itself from the environment
```
## Authentication Methods
## Authentication
### Universal Auth (Recommended)
Two ways, because the server has one login route:
```python
from hanzo_kms import KMSClient, ClientSettings, AuthenticationOptions, UniversalAuthMethod
| Setting | Behavior |
|---------|----------|
| `client_id` + `client_secret` | Exchanged at `POST /v1/kms/auth/login` for a bearer token |
| `access_token` | A pre-issued IAM bearer token, used as-is |
client = KMSClient(ClientSettings(
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id="...",
client_secret="...",
)
)
))
```
The AWS / Azure / GCP / Kubernetes / SRP methods this SDK used to advertise
were Infisical's. luxfi/kms has never served them.
### Kubernetes Auth
## CLI
For workloads running in Kubernetes:
```python
from hanzo_kms import KMSClient, ClientSettings, AuthenticationOptions, KubernetesAuthMethod
client = KMSClient(ClientSettings(
auth=AuthenticationOptions(
kubernetes=KubernetesAuthMethod(
identity_id="your-identity-id",
# Uses default service account token path
)
)
))
```
### AWS IAM Auth
```python
from hanzo_kms import KMSClient, ClientSettings, AuthenticationOptions, AWSIamAuthMethod
client = KMSClient(ClientSettings(
auth=AuthenticationOptions(
aws_iam=AWSIamAuthMethod(
identity_id="your-identity-id",
)
)
))
```bash
hanzo-kms list --path providers/lux --env prod
hanzo-kms get providers/lux deploy-mnemonic --env prod
hanzo-kms set providers/lux deploy-mnemonic "word word ..." --env prod
hanzo-kms delete providers/lux deploy-mnemonic --env prod
hanzo-kms export --path providers/lux --env prod --format json
```
## API Reference
@@ -126,19 +108,37 @@ client = KMSClient(ClientSettings(
| Method | Description |
|--------|-------------|
| `list_secrets(project_id, environment, path="/")` | List all secrets |
| `get_secret(project_id, environment, secret_name)` | Get a single secret |
| `get_value(project_id, environment, secret_name, default=None)` | Get just the value |
| `create_secret(project_id, environment, secret_name, secret_value)` | Create a secret |
| `update_secret(project_id, environment, secret_name, secret_value)` | Update a secret |
| `delete_secret(project_id, environment, secret_name)` | Delete a secret |
| `inject_env(project_id, environment, overwrite=False)` | Inject into os.environ |
| `list_secrets(path="", env="default")` | Secret **names** at a path |
| `get_secret(path, name, env="default")` | The secret's value |
| `put_secret(path, name, value, env="default")` | Create or replace |
| `delete_secret(path, name, env="default")` | Delete |
| `inject_env(path="", env="default", overwrite=False)` | Load a path into `os.environ` |
| `health()` | `{"service": "kms", "status": "ok"}` |
## Server surface
```
POST /v1/kms/auth/login {clientId, clientSecret} -> {accessToken, expiresIn}
GET /v1/kms/orgs/{org}/secrets?path=&env= -> {"names": [...]}
GET /v1/kms/orgs/{org}/secrets/{path}/{name}?env= -> {"secret": {"value": "..."}}
POST /v1/kms/orgs/{org}/secrets {path, name, env, value}
DELETE /v1/kms/orgs/{org}/secrets/{path}/{name}?env=
GET /healthz | /v1/kms/healthz -> {"service": "kms", "status": "ok"}
```
Two things follow from it:
- **The server splits the trailing path at its LAST slash** into `(path, name)`.
A name may not contain `/` — this SDK rejects one rather than let a write
land under a key the matching read can never find.
- **There is no versioned read.** KMS holds exactly one value per
(path, name, env), so `get_secret(..., version=N)` raises
`VersionUnsupportedError` instead of quietly handing back the current value.
## Compatibility
This SDK is compatible with:
- Hanzo KMS (https://kms.hanzo.ai)
- Lux KMS (https://kms.lux.network)
- Hanzo KMS — https://kms.hanzo.ai
- Lux KMS — https://kms.lux.network
## License
pkg/hanzo-kms/hanzo_kms/__init__.py
+26 -59
View File
@@ -1,79 +1,46 @@
"""
Hanzo KMS - Secret Management SDK for Python
"""Hanzo KMS - Secret Management SDK for Python
A pure Python SDK for Hanzo/Lux KMS.
A pure Python SDK for the canonical luxfi/kms surface (kms.hanzo.ai,
kms.lux.network). A secret is identified by (org, path, name, env).
Usage:
from hanzo_kms import KMSClient, ClientSettings, UniversalAuthMethod, AuthenticationOptions
from hanzo_kms import KMSClient, ClientSettings
client = KMSClient(ClientSettings(
site_url="https://kms.hanzo.ai",
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id="your-client-id",
client_secret="your-client-secret",
)
)
org="lux",
client_id="your-client-id",
client_secret="your-client-secret",
))
# List all secrets
secrets = client.list_secrets(project_id="my-project", environment="production")
# List the secret names at a path
names = client.list_secrets("providers/lux", env="prod")
# Get a specific secret
secret = client.get_secret(
project_id="my-project",
environment="production",
secret_name="DATABASE_URL"
)
print(secret.secret_value)
# Read one value
mnemonic = client.get_secret("providers/lux", "deploy-mnemonic", env="prod")
# Inject secrets into environment
client.inject_env(project_id="my-project", environment="production")
# Create or replace (one upsert — KMS holds one value per path/name/env)
client.put_secret("providers/lux", "deploy-mnemonic", mnemonic, env="prod")
# Load a whole path into os.environ
client.inject_env("providers/lux", env="prod")
With HANZO_KMS_ORG / HANZO_KMS_CLIENT_ID / HANZO_KMS_CLIENT_SECRET set,
`KMSClient()` configures itself.
"""
__version__ = "1.0.0"
__version__ = "1.1.1"
from .async_client import AsyncKMSClient
from .client import KMSClient
from .models import (
AuthenticationOptions,
AWSIamAuthMethod,
AzureAuthMethod,
ClientSettings,
CreateSecretOptions,
DeleteSecretOptions,
GCPIamAuthMethod,
GCPIDTokenAuthMethod,
GetSecretOptions,
KubernetesAuthMethod,
ListSecretsOptions,
SecretElement,
TokenAuthMethod,
UniversalAuthMethod,
UpdateSecretOptions,
UserPasswordAuthMethod,
)
from .models import ClientSettings, TokenResponse, settings_from_env
from .routes import DEFAULT_ENV, VersionUnsupportedError
__all__ = [
# Main clients
"KMSClient",
"AsyncKMSClient",
# Settings
"ClientSettings",
"AuthenticationOptions",
# Auth methods
"UniversalAuthMethod",
"AWSIamAuthMethod",
"AzureAuthMethod",
"GCPIamAuthMethod",
"GCPIDTokenAuthMethod",
"KubernetesAuthMethod",
# Options
"GetSecretOptions",
"ListSecretsOptions",
"CreateSecretOptions",
"UpdateSecretOptions",
"DeleteSecretOptions",
# Response types
"SecretElement",
"settings_from_env",
"TokenResponse",
"DEFAULT_ENV",
"VersionUnsupportedError",
]
pkg/hanzo-kms/hanzo_kms/async_client.py
+126 -259
View File
@@ -1,72 +1,49 @@
"""
Hanzo KMS Async Client - Async Python implementation
"""Hanzo KMS client — asynchronous.
An async-first Hanzo KMS client.
Mirror image of :class:`hanzo_kms.client.KMSClient`: same methods, same
arguments, same shared pure functions from :mod:`hanzo_kms.routes` — only the
I/O differs.
"""
import os
import time
from typing import Optional
from typing import Any, Optional
import httpx
from .models import (
AuthenticationOptions,
ClientSettings,
CreateSecretOptions,
DeleteSecretOptions,
GetSecretOptions,
ListSecretsOptions,
SecretElement,
SecretsResponse,
TokenResponse,
UpdateSecretOptions,
)
from . import routes
from .models import ClientSettings, TokenResponse, settings_from_env
class AsyncKMSClient:
"""
Async Hanzo KMS Client for secret management.
"""Async Hanzo KMS client for secret management.
Example:
async with AsyncKMSClient(settings) as client:
secrets = await client.list_secrets("myproject", "production")
async with AsyncKMSClient() as client:
names = await client.list_secrets("providers/lux", env="prod")
"""
def __init__(
self,
settings: Optional[ClientSettings] = None,
*,
debug: bool = False,
):
self.settings = settings or self._settings_from_env()
"""Initialize the client.
Args:
settings: Configuration; read from the environment when omitted.
debug: Enable debug logging.
"""
self.settings = settings or settings_from_env()
self.debug = debug
self._access_token: Optional[str] = None
self._token_expires_at: float = 0
self._access_token = ""
self._token_expires_at = 0.0
self._http_client: Optional[httpx.AsyncClient] = None
def _settings_from_env(self) -> ClientSettings:
"""Create settings from environment variables."""
from .models import UniversalAuthMethod
site_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
organization = os.getenv("HANZO_KMS_ORG", "hanzo")
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
auth = None
if client_id and client_secret:
auth = AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id=client_id,
client_secret=client_secret,
)
)
return ClientSettings(site_url=site_url, organization=organization, auth=auth)
@property
def http(self) -> httpx.AsyncClient:
"""Get or create async HTTP client."""
"""Get or create the async HTTP client."""
if self._http_client is None:
self._http_client = httpx.AsyncClient(
base_url=self.settings.site_url.rstrip("/"),
@@ -78,256 +55,146 @@ class AsyncKMSClient:
)
return self._http_client
async def _get_access_token(self) -> str:
"""Get valid access token, refreshing if needed."""
# =========================================================================
# Auth
# =========================================================================
async def _token(self) -> str:
"""Return a valid bearer token, logging in when needed."""
if self.settings.access_token:
return self.settings.access_token
if self._access_token and time.time() < self._token_expires_at - 60:
return self._access_token
auth = self.settings.auth
if not auth:
raise ValueError("No authentication configured")
# Universal Auth
if auth.universal_auth:
response = await self.http.post(
"/v1/kms/auth/login",
json={
"clientId": auth.universal_auth.client_id,
"clientSecret": auth.universal_auth.client_secret,
},
if not (self.settings.client_id and self.settings.client_secret):
raise ValueError(
"no KMS credentials: set access_token, or client_id and client_secret "
"(HANZO_KMS_TOKEN, or HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET)"
)
response.raise_for_status()
data = response.json()
token_data = TokenResponse.model_validate(data)
self._access_token = token_data.access_token
self._token_expires_at = time.time() + token_data.expires_in
return self._access_token
# Kubernetes Auth
if auth.kubernetes:
token_path = auth.kubernetes.service_account_token_path
if os.path.exists(token_path):
with open(token_path) as f:
k8s_token = f.read().strip()
response = await self.http.post(
routes.LOGIN,
json={
"clientId": self.settings.client_id,
"clientSecret": self.settings.client_secret,
},
)
response.raise_for_status()
token = TokenResponse.model_validate(response.json())
self._access_token = token.access_token
self._token_expires_at = time.time() + token.expires_in
return self._access_token
response = await self.http.post(
"/api/v1/auth/kubernetes-auth/login",
json={
"identityId": auth.kubernetes.identity_id,
"jwt": k8s_token,
},
)
response.raise_for_status()
data = response.json()
token_data = TokenResponse.model_validate(data)
self._access_token = token_data.access_token
self._token_expires_at = time.time() + token_data.expires_in
return self._access_token
async def _headers(self) -> dict[str, str]:
"""Bearer header. The org is carried by the URL and the JWT `owner`
claim — the server reads no org header."""
return {"Authorization": f"Bearer {await self._token()}"}
raise ValueError("No valid authentication method configured")
# =========================================================================
# Secrets
# =========================================================================
async def _auth_headers(self) -> dict[str, str]:
"""Get authorization headers including organization context."""
token = await self._get_access_token()
return {
"Authorization": f"Bearer {token}",
"X-Org-Name": self.settings.organization,
}
async def list_secrets(
self, path: str = "", env: str = routes.DEFAULT_ENV
) -> list[str]:
"""List the secret names stored at ``path``.
Names only: the server's list route returns ``{"names": [...]}`` with
no values, so reading a value takes a :meth:`get_secret` per name.
"""
response = await self.http.get(
routes.secrets_url(self.settings.org),
params=routes.list_params(path, env),
headers=await self._headers(),
)
response.raise_for_status()
return routes.names_of(response.json())
async def get_secret(
self,
project_id: str,
environment: str,
secret_name: str,
path: str = "/",
**kwargs,
) -> SecretElement:
"""Get a single secret by name."""
options = GetSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
path=path,
**kwargs,
)
path: str,
name: str,
env: str = routes.DEFAULT_ENV,
version: Optional[int] = None,
) -> str:
"""Read the value of one secret.
Args:
path: Secret path, e.g. ``"providers/lux"``.
name: Secret name, e.g. ``"deploy-mnemonic"``. May not contain ``/``.
env: Environment bucket.
version: Must be None — see :class:`~hanzo_kms.routes.VersionUnsupportedError`.
"""
routes.check_version(version)
response = await self.http.get(
f"/api/v3/secrets/raw/{options.secret_name}",
params={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"type": options.type,
},
headers=await self._auth_headers(),
routes.secret_url(self.settings.org, path, name),
params=routes.env_params(env),
headers=await self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
return routes.value_of(response.json())
async def list_secrets(
async def put_secret(
self,
project_id: str,
environment: str,
path: str = "/",
attach_to_process_env: bool = False,
**kwargs,
) -> list[SecretElement]:
"""List all secrets in a project/environment."""
options = ListSecretsOptions(
project_id=project_id,
environment=environment,
path=path,
attach_to_process_env=attach_to_process_env,
**kwargs,
)
response = await self.http.get(
"/api/v3/secrets/raw",
params={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"include_imports": str(options.include_imports).lower(),
"recursive": str(options.recursive).lower(),
"expandSecretReferences": str(options.expand_secret_references).lower(),
},
headers=await self._auth_headers(),
)
response.raise_for_status()
data = response.json()
secrets_data = SecretsResponse.model_validate(data)
if options.attach_to_process_env:
for secret in secrets_data.secrets:
if secret.secret_key not in os.environ:
os.environ[secret.secret_key] = secret.secret_value
return secrets_data.secrets
async def create_secret(
self,
project_id: str,
environment: str,
secret_name: str,
secret_value: str,
**kwargs,
) -> SecretElement:
"""Create a new secret."""
options = CreateSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
secret_value=secret_value,
**kwargs,
)
path: str,
name: str,
value: str,
env: str = routes.DEFAULT_ENV,
) -> None:
"""Create or replace a secret.
One upsert, not a create/update pair: luxfi/kms holds exactly one
value per (path, name, env) and a write replaces it in place.
"""
response = await self.http.post(
f"/api/v3/secrets/raw/{options.secret_name}",
json={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"secretValue": options.secret_value,
"secretComment": options.secret_comment,
"type": options.type,
},
headers=await self._auth_headers(),
routes.secrets_url(self.settings.org),
json=routes.upsert_body(path, name, value, env),
headers=await self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
async def update_secret(
self,
project_id: str,
environment: str,
secret_name: str,
secret_value: str,
**kwargs,
) -> SecretElement:
"""Update an existing secret."""
options = UpdateSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
secret_value=secret_value,
**kwargs,
)
response = await self.http.patch(
f"/api/v3/secrets/raw/{options.secret_name}",
json={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"secretValue": options.secret_value,
},
headers=await self._auth_headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
async def delete_secret(
self,
project_id: str,
environment: str,
secret_name: str,
**kwargs,
) -> SecretElement:
path: str,
name: str,
env: str = routes.DEFAULT_ENV,
) -> None:
"""Delete a secret."""
options = DeleteSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
**kwargs,
)
response = await self.http.request(
"DELETE",
f"/api/v3/secrets/raw/{options.secret_name}",
json={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
},
headers=await self._auth_headers(),
response = await self.http.delete(
routes.secret_url(self.settings.org, path, name),
params=routes.env_params(env),
headers=await self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
async def health(self) -> dict[str, Any]:
"""Probe the server: ``{"service": "kms", "status": "ok"}``. No auth."""
response = await self.http.get(routes.HEALTH)
response.raise_for_status()
return response.json()
async def inject_env(
self,
project_id: str,
environment: str,
path: str = "/",
path: str = "",
env: str = routes.DEFAULT_ENV,
overwrite: bool = False,
) -> int:
"""Inject all secrets into environment variables."""
secrets = await self.list_secrets(project_id, environment, path)
"""Load every secret at ``path`` into ``os.environ``, keyed by name.
Costs one list request plus one read per secret — the list route
returns names, not values.
Returns:
The number of variables set.
"""
count = 0
for secret in secrets:
if overwrite or secret.secret_key not in os.environ:
os.environ[secret.secret_key] = secret.secret_value
for name in await self.list_secrets(path, env):
if overwrite or name not in os.environ:
os.environ[name] = await self.get_secret(path, name, env)
count += 1
return count
async def get_value(
self,
project_id: str,
environment: str,
secret_name: str,
default: Optional[str] = None,
) -> Optional[str]:
"""Get just the value of a secret."""
try:
secret = await self.get_secret(project_id, environment, secret_name)
return secret.secret_value
except httpx.HTTPStatusError:
return default
async def close(self) -> None:
"""Close the HTTP client."""
if self._http_client:
@@ -337,5 +204,5 @@ class AsyncKMSClient:
async def __aenter__(self) -> "AsyncKMSClient":
return self
async def __aexit__(self, *args) -> None:
async def __aexit__(self, *args: Any) -> None:
await self.close()
pkg/hanzo-kms/hanzo_kms/cli.py
+82 -107
View File
@@ -1,103 +1,89 @@
"""
Hanzo KMS CLI — fetch secrets non-interactively.
"""Hanzo KMS CLI — read and write secrets non-interactively.
A secret is (org, path, name, env). The org comes from ``--org`` /
``HANZO_KMS_ORG`` (default ``hanzo``); path and name are positional.
Usage:
# Via env vars
HANZO_KMS_[email protected] HANZO_KMS_PASSWORD=... python -m hanzo_kms get SECRET_NAME --project ID --env prod
# Machine identity from the environment
export HANZO_KMS_CLIENT_ID=... HANZO_KMS_CLIENT_SECRET=...
python -m hanzo_kms get providers/lux deploy-mnemonic --env prod
# Via flags
python -m hanzo_kms get SECRET_NAME --email [email protected] --password ... --project ID --env prod
# Or a pre-issued IAM bearer token
HANZO_KMS_TOKEN=... python -m hanzo_kms list --path providers/lux --env prod
# List secrets
python -m hanzo_kms list --project ID --env prod
# With universal auth
HANZO_KMS_CLIENT_ID=... HANZO_KMS_CLIENT_SECRET=... python -m hanzo_kms get SECRET_NAME --project ID --env prod
# With pre-authenticated token
HANZO_KMS_TOKEN=... python -m hanzo_kms get SECRET_NAME --project ID --env prod
python -m hanzo_kms set providers/lux deploy-mnemonic "word word ..." --env prod
python -m hanzo_kms delete providers/lux deploy-mnemonic --env prod
python -m hanzo_kms export --path providers/lux --env prod --format json
"""
import argparse
import os
import json
import sys
from .client import KMSClient
from .models import (
AuthenticationOptions,
ClientSettings,
TokenAuthMethod,
UniversalAuthMethod,
UserPasswordAuthMethod,
)
from .models import settings_from_env
from .routes import DEFAULT_ENV
def _build_client(args: argparse.Namespace) -> KMSClient:
"""Build KMS client from CLI args + env vars."""
site_url = args.url or os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
def _client(args: argparse.Namespace) -> KMSClient:
"""Build a client from the environment, overlaid with CLI flags."""
settings = settings_from_env()
if args.url:
settings.site_url = args.url
if args.org:
settings.org = args.org
if args.token:
settings.access_token = args.token
if args.client_id:
settings.client_id = args.client_id
if args.client_secret:
settings.client_secret = args.client_secret
# Auth priority: token > universal-auth > user/password > env
token = args.token or os.getenv("HANZO_KMS_TOKEN", "")
client_id = args.client_id or os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = args.client_secret or os.getenv("HANZO_KMS_CLIENT_SECRET", "")
email = args.email or os.getenv("HANZO_KMS_EMAIL", "")
password = args.password or os.getenv("HANZO_KMS_PASSWORD", "")
auth = None
if token:
auth = AuthenticationOptions(token=TokenAuthMethod(access_token=token))
elif client_id and client_secret:
auth = AuthenticationOptions(
universal_auth=UniversalAuthMethod(client_id=client_id, client_secret=client_secret)
if not (settings.access_token or (settings.client_id and settings.client_secret)):
print(
"Error: no auth configured. Set HANZO_KMS_TOKEN, or "
"HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET.",
file=sys.stderr,
)
elif email and password:
auth = AuthenticationOptions(
user_password=UserPasswordAuthMethod(email=email, password=password)
)
if not auth:
# Try env fallback
client = KMSClient(debug=getattr(args, "debug", False))
if client.settings.auth:
return client
print("Error: No auth configured. Set HANZO_KMS_TOKEN, HANZO_KMS_CLIENT_ID/SECRET, or HANZO_KMS_EMAIL/PASSWORD", file=sys.stderr)
sys.exit(1)
settings = ClientSettings(site_url=site_url, auth=auth)
return KMSClient(settings=settings, debug=getattr(args, "debug", False))
return KMSClient(settings, debug=args.debug)
def main() -> None:
parser = argparse.ArgumentParser(prog="hanzo-kms", description="Hanzo KMS CLI")
parser.add_argument("--url", help="KMS URL (default: https://kms.hanzo.ai)")
parser.add_argument("--token", help="Pre-authenticated access token")
parser.add_argument("--client-id", help="Universal auth client ID")
parser.add_argument("--client-secret", help="Universal auth client secret")
parser.add_argument("--email", help="User email for SRP auth")
parser.add_argument("--password", help="User password for SRP auth")
parser.add_argument("--org", help="Organization (default: hanzo)")
parser.add_argument("--token", help="Pre-issued IAM bearer token")
parser.add_argument("--client-id", help="Machine identity client ID")
parser.add_argument("--client-secret", help="Machine identity client secret")
parser.add_argument("--debug", action="store_true")
sub = parser.add_subparsers(dest="command")
# get command
get_p = sub.add_parser("get", help="Get a secret value")
list_p = sub.add_parser("list", help="List secret names at a path")
list_p.add_argument("--path", default="", help="Secret path")
list_p.add_argument("--env", default=DEFAULT_ENV, help=f"Environment (default: {DEFAULT_ENV})")
get_p = sub.add_parser("get", help="Print a secret value")
get_p.add_argument("path", help="Secret path, e.g. providers/lux")
get_p.add_argument("name", help="Secret name")
get_p.add_argument("--project", required=True, help="Project ID")
get_p.add_argument("--env", default="prod", help="Environment (default: prod)")
get_p.add_argument("--path", default="/", help="Secret path")
get_p.add_argument("--env", default=DEFAULT_ENV, help=f"Environment (default: {DEFAULT_ENV})")
# list command
list_p = sub.add_parser("list", help="List secrets")
list_p.add_argument("--project", required=True, help="Project ID")
list_p.add_argument("--env", default="prod", help="Environment (default: prod)")
list_p.add_argument("--path", default="/", help="Secret path")
list_p.add_argument("--keys-only", action="store_true", help="Only print keys")
set_p = sub.add_parser("set", help="Create or replace a secret")
set_p.add_argument("path", help="Secret path")
set_p.add_argument("name", help="Secret name")
set_p.add_argument("value", help="Secret value")
set_p.add_argument("--env", default=DEFAULT_ENV, help=f"Environment (default: {DEFAULT_ENV})")
# export command
export_p = sub.add_parser("export", help="Export secrets as env vars")
export_p.add_argument("--project", required=True, help="Project ID")
export_p.add_argument("--env", default="prod", help="Environment (default: prod)")
export_p.add_argument("--path", default="/", help="Secret path")
del_p = sub.add_parser("delete", help="Delete a secret")
del_p.add_argument("path", help="Secret path")
del_p.add_argument("name", help="Secret name")
del_p.add_argument("--env", default=DEFAULT_ENV, help=f"Environment (default: {DEFAULT_ENV})")
export_p = sub.add_parser("export", help="Export every secret at a path")
export_p.add_argument("--path", default="", help="Secret path")
export_p.add_argument("--env", default=DEFAULT_ENV, help=f"Environment (default: {DEFAULT_ENV})")
export_p.add_argument("--format", choices=["env", "json"], default="env")
args = parser.parse_args()
@@ -105,44 +91,33 @@ def main() -> None:
parser.print_help()
sys.exit(1)
client = _build_client(args)
client = _client(args)
try:
if args.command == "list":
for name in client.list_secrets(args.path, args.env):
print(name)
if args.command == "get":
secret = client.get_secret(
project_id=args.project,
environment=args.env,
secret_name=args.name,
path=args.path,
)
print(secret.secret_value)
elif args.command == "get":
print(client.get_secret(args.path, args.name, args.env))
elif args.command == "list":
secrets = client.list_secrets(
project_id=args.project,
environment=args.env,
path=args.path,
)
for s in secrets:
if args.keys_only:
print(s.secret_key)
elif args.command == "set":
client.put_secret(args.path, args.name, args.value, args.env)
print(f"set {args.path}/{args.name} ({args.env})")
elif args.command == "delete":
client.delete_secret(args.path, args.name, args.env)
print(f"deleted {args.path}/{args.name} ({args.env})")
elif args.command == "export":
names = client.list_secrets(args.path, args.env)
values = {name: client.get_secret(args.path, name, args.env) for name in names}
if args.format == "json":
print(json.dumps(values, indent=2))
else:
print(f"{s.secret_key}={s.secret_value}")
elif args.command == "export":
import json as _json
secrets = client.list_secrets(
project_id=args.project,
environment=args.env,
path=args.path,
)
if args.format == "json":
print(_json.dumps({s.secret_key: s.secret_value for s in secrets}, indent=2))
else:
for s in secrets:
print(f"{s.secret_key}={s.secret_value}")
client.close()
for name, value in values.items():
print(f"{name}={value}")
finally:
client.close()
if __name__ == "__main__":
pkg/hanzo-kms/hanzo_kms/client.py
+119 -438
View File
@@ -1,108 +1,61 @@
"""
Hanzo KMS Client - Pure Python implementation
"""Hanzo KMS client — synchronous.
A full-featured Hanzo KMS client.
Speaks the canonical luxfi/kms surface; see :mod:`hanzo_kms.routes` for the
route table and for why the old ``/api/*`` paths never worked. This is the
mirror image of :class:`hanzo_kms.async_client.AsyncKMSClient`: same methods,
same arguments, same shared pure functions — only the I/O differs.
"""
import os
import time
from typing import Optional
from typing import Any, Optional
import httpx
from .models import (
AuthenticationOptions,
ClientSettings,
CreateSecretOptions,
DeleteSecretOptions,
GetSecretOptions,
ListSecretsOptions,
SecretElement,
SecretsResponse,
TokenResponse,
UpdateSecretOptions,
)
from . import routes
from .models import ClientSettings, TokenResponse, settings_from_env
class KMSClient:
"""
Hanzo KMS Client for secret management.
"""Hanzo KMS client for secret management.
Supports multiple authentication methods:
- Universal Auth (client_id/client_secret)
- AWS IAM
- Azure AD
- GCP IAM
- Kubernetes Service Account
A secret is identified by (org, path, name, env). ``org`` comes from
settings — it scopes both the URL and the JWT — and the rest are
per-call.
Example:
client = KMSClient(ClientSettings(
site_url="https://kms.hanzo.ai",
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id="your-client-id",
client_secret="your-client-secret",
)
)
org="lux",
client_id="...",
client_secret="...",
))
mnemonic = client.get_secret("providers/lux", "deploy-mnemonic", env="prod")
secrets = client.list_secrets(project_id="myproject", environment="production")
With ``HANZO_KMS_ORG`` / ``HANZO_KMS_CLIENT_ID`` / ``HANZO_KMS_CLIENT_SECRET``
set, ``KMSClient()`` configures itself.
"""
def __init__(
self,
settings: Optional[ClientSettings] = None,
*,
debug: bool = False,
):
"""Initialize KMS client.
"""Initialize the client.
Args:
settings: Client configuration (defaults from environment if not provided)
debug: Enable debug logging
settings: Configuration; read from the environment when omitted.
debug: Enable debug logging.
"""
self.settings = settings or self._settings_from_env()
self.settings = settings or settings_from_env()
self.debug = debug
self._access_token: Optional[str] = None
self._token_expires_at: float = 0
self._access_token = ""
self._token_expires_at = 0.0
self._http_client: Optional[httpx.Client] = None
def _settings_from_env(self) -> ClientSettings:
"""Create settings from environment variables."""
from .models import TokenAuthMethod, UniversalAuthMethod, UserPasswordAuthMethod
site_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
organization = os.getenv("HANZO_KMS_ORG", "hanzo")
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
email = os.getenv("HANZO_KMS_EMAIL", "")
password = os.getenv("HANZO_KMS_PASSWORD", "")
token = os.getenv("HANZO_KMS_TOKEN", "")
auth = None
if token:
auth = AuthenticationOptions(
token=TokenAuthMethod(access_token=token)
)
elif client_id and client_secret:
auth = AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id=client_id,
client_secret=client_secret,
)
)
elif email and password:
auth = AuthenticationOptions(
user_password=UserPasswordAuthMethod(
email=email,
password=password,
)
)
return ClientSettings(site_url=site_url, organization=organization, auth=auth)
@property
def http(self) -> httpx.Client:
"""Get or create HTTP client."""
"""Get or create the HTTP client."""
if self._http_client is None:
self._http_client = httpx.Client(
base_url=self.settings.site_url.rstrip("/"),
@@ -114,415 +67,144 @@ class KMSClient:
)
return self._http_client
def _user_login(self, email: str, password: str) -> str:
"""Login with email/password via v3 non-SRP endpoint."""
response = self.http.post(
"/api/v3/auth/login",
json={"email": email, "password": password},
)
response.raise_for_status()
data = response.json()
return data.get("accessToken", data.get("access_token", ""))
# =========================================================================
# Auth
# =========================================================================
def _token(self) -> str:
"""Return a valid bearer token, logging in when needed."""
if self.settings.access_token:
return self.settings.access_token
def _get_access_token(self) -> str:
"""Get valid access token, refreshing if needed."""
if self._access_token and time.time() < self._token_expires_at - 60:
return self._access_token
auth = self.settings.auth
if not auth:
raise ValueError("No authentication configured")
# Direct token auth
if auth.token:
self._access_token = auth.token.access_token
self._token_expires_at = time.time() + 86400 # assume 24h
return self._access_token
# User/password auth (v3 non-SRP)
if auth.user_password:
token = self._user_login(auth.user_password.email, auth.user_password.password)
self._access_token = token
self._token_expires_at = time.time() + 7200 # 2h
return self._access_token
# Universal Auth
if auth.universal_auth:
response = self.http.post(
"/v1/kms/auth/login",
json={
"clientId": auth.universal_auth.client_id,
"clientSecret": auth.universal_auth.client_secret,
},
if not (self.settings.client_id and self.settings.client_secret):
raise ValueError(
"no KMS credentials: set access_token, or client_id and client_secret "
"(HANZO_KMS_TOKEN, or HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET)"
)
response.raise_for_status()
data = response.json()
token_data = TokenResponse.model_validate(data)
self._access_token = token_data.access_token
self._token_expires_at = time.time() + token_data.expires_in
return self._access_token
# Kubernetes Auth
if auth.kubernetes:
token_path = auth.kubernetes.service_account_token_path
if os.path.exists(token_path):
with open(token_path) as f:
k8s_token = f.read().strip()
response = self.http.post(
routes.LOGIN,
json={
"clientId": self.settings.client_id,
"clientSecret": self.settings.client_secret,
},
)
response.raise_for_status()
token = TokenResponse.model_validate(response.json())
self._access_token = token.access_token
self._token_expires_at = time.time() + token.expires_in
return self._access_token
response = self.http.post(
"/api/v1/auth/kubernetes-auth/login",
json={
"identityId": auth.kubernetes.identity_id,
"jwt": k8s_token,
},
)
response.raise_for_status()
data = response.json()
token_data = TokenResponse.model_validate(data)
self._access_token = token_data.access_token
self._token_expires_at = time.time() + token_data.expires_in
return self._access_token
def _headers(self) -> dict[str, str]:
"""Bearer header. The org is carried by the URL and the JWT `owner`
claim — the server reads no org header."""
return {"Authorization": f"Bearer {self._token()}"}
# AWS IAM Auth
if auth.aws_iam:
# Get AWS credentials from environment/instance metadata
import json
# =========================================================================
# Secrets
# =========================================================================
try:
import boto3
def list_secrets(self, path: str = "", env: str = routes.DEFAULT_ENV) -> list[str]:
"""List the secret names stored at ``path``.
session = boto3.Session()
credentials = session.get_credentials()
_region = session.region_name or "us-east-1"
response = self.http.post(
"/api/v1/auth/aws-auth/login",
json={
"identityId": auth.aws_iam.identity_id,
"iamHttpRequestMethod": "POST",
"iamRequestBody": "",
"iamRequestHeaders": json.dumps(
{
"X-Amz-Date": credentials.token or "",
}
),
},
)
response.raise_for_status()
data = response.json()
token_data = TokenResponse.model_validate(data)
self._access_token = token_data.access_token
self._token_expires_at = time.time() + token_data.expires_in
return self._access_token
except ImportError as e:
raise ValueError(
"boto3 required for AWS IAM auth: pip install boto3"
) from e
raise ValueError("No valid authentication method configured")
def _auth_headers(self) -> dict[str, str]:
"""Get authorization headers including organization context."""
token = self._get_access_token()
return {
"Authorization": f"Bearer {token}",
"X-Org-Name": self.settings.organization,
}
def _resolve_project_id(self, project_id: str) -> tuple[str, str]:
"""Resolve org-scoped project ID.
Supports formats:
- "project" -> uses settings.organization
- "org/project" -> uses specified org
Returns:
Tuple of (organization, project_id)
Names only: the server's list route returns ``{"names": [...]}`` with
no values, so reading a value takes a :meth:`get_secret` per name.
"""
if "/" in project_id:
org, proj = project_id.split("/", 1)
return (org, proj)
return (self.settings.organization, project_id)
# =========================================================================
# Secret Operations
# =========================================================================
response = self.http.get(
routes.secrets_url(self.settings.org),
params=routes.list_params(path, env),
headers=self._headers(),
)
response.raise_for_status()
return routes.names_of(response.json())
def get_secret(
self,
project_id: str,
environment: str,
secret_name: str,
path: str = "/",
**kwargs,
) -> SecretElement:
"""Get a single secret by name.
path: str,
name: str,
env: str = routes.DEFAULT_ENV,
version: Optional[int] = None,
) -> str:
"""Read the value of one secret.
Args:
project_id: Project ID or slug
environment: Environment slug (e.g., "production")
secret_name: Name of the secret
path: Secret path (default "/")
Returns:
SecretElement with the secret data
path: Secret path, e.g. ``"providers/lux"``.
name: Secret name, e.g. ``"deploy-mnemonic"``. May not contain ``/``.
env: Environment bucket.
version: Must be None — see :class:`~hanzo_kms.routes.VersionUnsupportedError`.
"""
options = GetSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
path=path,
**kwargs,
)
routes.check_version(version)
response = self.http.get(
f"/api/v3/secrets/raw/{options.secret_name}",
params={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"type": options.type,
"include_imports": str(options.include_imports).lower(),
},
headers=self._auth_headers(),
routes.secret_url(self.settings.org, path, name),
params=routes.env_params(env),
headers=self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
return routes.value_of(response.json())
def list_secrets(
def put_secret(
self,
project_id: str,
environment: str,
path: str = "/",
attach_to_process_env: bool = False,
**kwargs,
) -> list[SecretElement]:
"""List all secrets in a project/environment.
path: str,
name: str,
value: str,
env: str = routes.DEFAULT_ENV,
) -> None:
"""Create or replace a secret.
Args:
project_id: Project ID or slug
environment: Environment slug
path: Secret path (default "/")
attach_to_process_env: If True, set secrets as environment variables
Returns:
List of SecretElement
One upsert, not a create/update pair: luxfi/kms holds exactly one
value per (path, name, env) and a write replaces it in place.
"""
options = ListSecretsOptions(
project_id=project_id,
environment=environment,
path=path,
attach_to_process_env=attach_to_process_env,
**kwargs,
)
response = self.http.get(
"/api/v3/secrets/raw",
params={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"include_imports": str(options.include_imports).lower(),
"recursive": str(options.recursive).lower(),
"expandSecretReferences": str(options.expand_secret_references).lower(),
},
headers=self._auth_headers(),
)
response.raise_for_status()
data = response.json()
secrets_data = SecretsResponse.model_validate(data)
# Optionally inject into environment
if options.attach_to_process_env:
for secret in secrets_data.secrets:
if secret.secret_key not in os.environ:
os.environ[secret.secret_key] = secret.secret_value
return secrets_data.secrets
def create_secret(
self,
project_id: str,
environment: str,
secret_name: str,
secret_value: str,
**kwargs,
) -> SecretElement:
"""Create a new secret.
Args:
project_id: Project ID or slug
environment: Environment slug
secret_name: Name of the secret
secret_value: Value of the secret
Returns:
Created SecretElement
"""
options = CreateSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
secret_value=secret_value,
**kwargs,
)
payload: dict = {
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"secretValue": options.secret_value,
"type": options.type,
}
if options.secret_comment is not None:
payload["secretComment"] = options.secret_comment
response = self.http.post(
f"/api/v3/secrets/raw/{options.secret_name}",
json=payload,
headers=self._auth_headers(),
routes.secrets_url(self.settings.org),
json=routes.upsert_body(path, name, value, env),
headers=self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
def update_secret(
self,
project_id: str,
environment: str,
secret_name: str,
secret_value: str,
**kwargs,
) -> SecretElement:
"""Update an existing secret.
Args:
project_id: Project ID or slug
environment: Environment slug
secret_name: Name of the secret
secret_value: New value
Returns:
Updated SecretElement
"""
options = UpdateSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
secret_value=secret_value,
**kwargs,
)
response = self.http.patch(
f"/api/v3/secrets/raw/{options.secret_name}",
json={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"secretValue": options.secret_value,
"secretComment": options.secret_comment,
"type": options.type,
},
headers=self._auth_headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
def delete_secret(
self,
project_id: str,
environment: str,
secret_name: str,
**kwargs,
) -> SecretElement:
"""Delete a secret.
Args:
project_id: Project ID or slug
environment: Environment slug
secret_name: Name of the secret
Returns:
Deleted SecretElement
"""
options = DeleteSecretOptions(
project_id=project_id,
environment=environment,
secret_name=secret_name,
**kwargs,
)
response = self.http.request(
"DELETE",
f"/api/v3/secrets/raw/{options.secret_name}",
json={
"workspaceId": options.project_id,
"environment": options.environment,
"secretPath": options.path,
"type": options.type,
},
headers=self._auth_headers(),
path: str,
name: str,
env: str = routes.DEFAULT_ENV,
) -> None:
"""Delete a secret."""
response = self.http.delete(
routes.secret_url(self.settings.org, path, name),
params=routes.env_params(env),
headers=self._headers(),
)
response.raise_for_status()
data = response.json()
return SecretElement.model_validate(data.get("secret", data))
# =========================================================================
# Convenience Methods
# =========================================================================
def health(self) -> dict[str, Any]:
"""Probe the server: ``{"service": "kms", "status": "ok"}``. No auth."""
response = self.http.get(routes.HEALTH)
response.raise_for_status()
return response.json()
def inject_env(
self,
project_id: str,
environment: str,
path: str = "/",
path: str = "",
env: str = routes.DEFAULT_ENV,
overwrite: bool = False,
) -> int:
"""Inject all secrets into environment variables.
"""Load every secret at ``path`` into ``os.environ``, keyed by name.
Args:
project_id: Project ID or slug
environment: Environment slug
path: Secret path
overwrite: If True, overwrite existing env vars
Costs one list request plus one read per secret — the list route
returns names, not values.
Returns:
Number of secrets injected
The number of variables set.
"""
secrets = self.list_secrets(project_id, environment, path)
count = 0
for secret in secrets:
if overwrite or secret.secret_key not in os.environ:
os.environ[secret.secret_key] = secret.secret_value
for name in self.list_secrets(path, env):
if overwrite or name not in os.environ:
os.environ[name] = self.get_secret(path, name, env)
count += 1
return count
def get_value(
self,
project_id: str,
environment: str,
secret_name: str,
default: Optional[str] = None,
) -> Optional[str]:
"""Get just the value of a secret.
Args:
project_id: Project ID or slug
environment: Environment slug
secret_name: Name of the secret
default: Default value if secret not found
Returns:
Secret value or default
"""
try:
secret = self.get_secret(project_id, environment, secret_name)
return secret.secret_value
except httpx.HTTPStatusError:
return default
def close(self) -> None:
"""Close the HTTP client."""
if self._http_client:
@@ -532,6 +214,5 @@ class KMSClient:
def __enter__(self) -> "KMSClient":
return self
def __exit__(self, *args) -> None:
def __exit__(self, *args: Any) -> None:
self.close()
pkg/hanzo-kms/hanzo_kms/models.py
+31 -203
View File
@@ -1,222 +1,50 @@
"""
Hanzo KMS - Data models for the SDK
"""Settings and response models for the canonical luxfi/kms surface."""
Pydantic models for API requests and responses.
"""
import os
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, Field
# =============================================================================
# Authentication Models
# =============================================================================
class UniversalAuthMethod(BaseModel):
"""Universal authentication using client credentials."""
client_id: str = Field(..., description="Client ID from KMS")
client_secret: str = Field(..., description="Client secret from KMS")
class AWSIamAuthMethod(BaseModel):
"""AWS IAM authentication."""
identity_id: str = Field(..., description="Identity ID in KMS")
class AzureAuthMethod(BaseModel):
"""Azure AD authentication."""
identity_id: str = Field(..., description="Identity ID in KMS")
resource: Optional[str] = Field(None, description="Azure resource")
class GCPIamAuthMethod(BaseModel):
"""GCP IAM authentication."""
identity_id: str = Field(..., description="Identity ID in KMS")
service_account_key_file_path: str = Field(
..., description="Path to service account key"
)
class GCPIDTokenAuthMethod(BaseModel):
"""GCP ID Token authentication."""
identity_id: str = Field(..., description="Identity ID in KMS")
class KubernetesAuthMethod(BaseModel):
"""Kubernetes service account authentication."""
identity_id: str = Field(..., description="Identity ID in KMS")
service_account_token_path: str = Field(
"/var/run/secrets/kubernetes.io/serviceaccount/token",
description="Path to service account token",
)
class UserPasswordAuthMethod(BaseModel):
"""User email/password authentication (SRP)."""
email: str = Field(..., description="User email")
password: str = Field(..., description="User password")
class TokenAuthMethod(BaseModel):
"""Direct token authentication (pre-authenticated)."""
access_token: str = Field(..., description="Pre-authenticated access token")
class AuthenticationOptions(BaseModel):
"""Authentication configuration - use one method."""
universal_auth: Optional[UniversalAuthMethod] = None
user_password: Optional[UserPasswordAuthMethod] = None
token: Optional[TokenAuthMethod] = None
aws_iam: Optional[AWSIamAuthMethod] = None
azure: Optional[AzureAuthMethod] = None
gcp_iam: Optional[GCPIamAuthMethod] = None
gcp_id_token: Optional[GCPIDTokenAuthMethod] = None
kubernetes: Optional[KubernetesAuthMethod] = None
# =============================================================================
# Client Settings
# =============================================================================
DEFAULT_SITE_URL = "https://kms.hanzo.ai"
DEFAULT_ORG = "hanzo"
class ClientSettings(BaseModel):
"""Client configuration settings."""
"""KMS client configuration.
site_url: str = Field("https://kms.hanzo.ai", description="KMS API URL")
organization: str = Field(
"hanzo", description="Organization name for multi-tenancy"
)
auth: Optional[AuthenticationOptions] = Field(
None, description="Authentication options"
Auth is one of exactly two things: a pre-issued IAM bearer token
(``access_token``), or machine-identity client credentials that
``POST /v1/kms/auth/login`` exchanges for one. luxfi/kms has a single
login route — the AWS / Azure / GCP / Kubernetes / SRP methods this SDK
used to model were Infisical's, and none of them were ever served.
"""
site_url: str = Field(DEFAULT_SITE_URL, description="KMS base URL")
org: str = Field(
DEFAULT_ORG,
description="Organization — scopes both the secret path and the JWT `owner` claim",
)
client_id: str = Field("", description="Machine identity client ID")
client_secret: str = Field("", description="Machine identity client secret")
access_token: str = Field("", description="Pre-issued IAM bearer token; skips login")
user_agent: str = Field("hanzo-kms-python", description="User agent string")
cache_ttl: int = Field(300, description="Cache TTL in seconds")
# =============================================================================
# Secret Models
# =============================================================================
class SecretElement(BaseModel):
"""A secret from KMS."""
id: str = Field(..., description="Secret ID")
secret_key: str = Field(..., alias="secretKey", description="Secret key/name")
secret_value: str = Field(..., alias="secretValue", description="Secret value")
secret_comment: Optional[str] = Field(
None, alias="secretComment", description="Comment"
def settings_from_env() -> ClientSettings:
"""Build settings from the ``HANZO_KMS_*`` environment variables."""
return ClientSettings(
site_url=os.getenv("HANZO_KMS_URL", DEFAULT_SITE_URL),
org=os.getenv("HANZO_KMS_ORG", DEFAULT_ORG),
client_id=os.getenv("HANZO_KMS_CLIENT_ID", ""),
client_secret=os.getenv("HANZO_KMS_CLIENT_SECRET", ""),
access_token=os.getenv("HANZO_KMS_TOKEN", ""),
)
version: int = Field(1, description="Secret version")
type: str = Field("shared", description="Secret type")
environment: str = Field(..., description="Environment slug")
workspace: str = Field(..., description="Workspace/project ID")
created_at: Optional[datetime] = Field(None, alias="createdAt")
updated_at: Optional[datetime] = Field(None, alias="updatedAt")
class Config:
populate_by_name = True
# =============================================================================
# Request Options
# =============================================================================
class GetSecretOptions(BaseModel):
"""Options for getting a single secret."""
project_id: str = Field(..., description="Project ID or slug")
environment: str = Field(..., description="Environment slug")
secret_name: str = Field(..., description="Secret key/name")
path: str = Field("/", description="Secret path")
type: str = Field("shared", description="Secret type")
include_imports: bool = Field(True, description="Include imported secrets")
class ListSecretsOptions(BaseModel):
"""Options for listing secrets."""
project_id: str = Field(..., description="Project ID or slug")
environment: str = Field(..., description="Environment slug")
path: str = Field("/", description="Secret path")
include_imports: bool = Field(True, description="Include imported secrets")
recursive: bool = Field(False, description="Recursively fetch from subpaths")
expand_secret_references: bool = Field(True, description="Expand ${} references")
attach_to_process_env: bool = Field(
False, description="Set as environment variables"
)
class CreateSecretOptions(BaseModel):
"""Options for creating a secret."""
project_id: str = Field(..., description="Project ID or slug")
environment: str = Field(..., description="Environment slug")
secret_name: str = Field(..., description="Secret key/name")
secret_value: str = Field(..., description="Secret value")
secret_comment: Optional[str] = Field(None, description="Comment")
path: str = Field("/", description="Secret path")
type: str = Field("shared", description="Secret type")
class UpdateSecretOptions(BaseModel):
"""Options for updating a secret."""
project_id: str = Field(..., description="Project ID or slug")
environment: str = Field(..., description="Environment slug")
secret_name: str = Field(..., description="Secret key/name")
secret_value: str = Field(..., description="New secret value")
secret_comment: Optional[str] = Field(None, description="Comment")
path: str = Field("/", description="Secret path")
type: str = Field("shared", description="Secret type")
class DeleteSecretOptions(BaseModel):
"""Options for deleting a secret."""
project_id: str = Field(..., description="Project ID or slug")
environment: str = Field(..., description="Environment slug")
secret_name: str = Field(..., description="Secret key/name")
path: str = Field("/", description="Secret path")
type: str = Field("shared", description="Secret type")
# =============================================================================
# Response Models
# =============================================================================
class TokenResponse(BaseModel):
"""Access token response from authentication."""
"""``POST /v1/kms/auth/login`` response."""
model_config = ConfigDict(populate_by_name=True)
access_token: str = Field(..., alias="accessToken")
expires_in: int = Field(..., alias="expiresIn")
token_type: str = Field("Bearer", alias="tokenType")
class Config:
populate_by_name = True
class SecretsResponse(BaseModel):
"""Response containing list of secrets."""
secrets: list[SecretElement]
imports: Optional[list[Any]] = None
class SecretResponse(BaseModel):
"""Response containing a single secret."""
secret: SecretElement
pkg/hanzo-kms/hanzo_kms/routes.py
+148
View File
@@ -0,0 +1,148 @@
"""The canonical luxfi/kms HTTP surface.
kms.hanzo.ai runs luxfi/kms. Its entire secret API is the routes below, all
under ``/v1/kms``::
POST /v1/kms/auth/login {clientId, clientSecret} -> {accessToken, expiresIn}
GET /v1/kms/orgs/{org}/secrets?path=&env= -> {"names": [...]}
GET /v1/kms/orgs/{org}/secrets/{path}/{name}?env= -> {"secret": {"value": "..."}}
POST /v1/kms/orgs/{org}/secrets {path, name, env, value} (create AND replace)
DELETE /v1/kms/orgs/{org}/secrets/{path}/{name}?env=
GET /healthz | /v1/kms/healthz -> {"service": "kms", "status": "ok"}
There is no ``/api/*`` surface and there never was. The Infisical paths this
SDK used to send (``/api/v3/secrets/raw``, ``/api/v3/auth/login``,
``/api/v1/auth/kubernetes-auth/login``) have always 404'd against luxfi/kms.
The break stayed invisible because older builds embedded a console SPA behind
a root catch-all that answered every unmatched path with ``200 text/html`` —
so a wrong URL surfaced as a JSON decode error rather than a 404, and read
like a parsing bug. That catch-all is gone (``cmd/kms/main.go`` now ends in
``notFoundJSON``), so wrong paths answer honest JSON 404s.
Everything here is a pure function: the sync and async clients stay mirror
images by sharing it, and one set of tests pins the wire shape for both.
"""
from typing import Any, Optional
from urllib.parse import quote
#: The server treats a read with no ``env`` as this bucket. Writes have no
#: default at all — the server 400s on an empty ``env`` rather than silently
#: committing to a bucket that project/env/path readers never resolve — so
#: this SDK always sends an explicit value.
DEFAULT_ENV = "default"
LOGIN = "/v1/kms/auth/login"
HEALTH = "/v1/kms/healthz"
class VersionUnsupportedError(ValueError):
"""Raised when a caller asks for a specific version of a secret.
luxfi/kms stores exactly one value per (path, name, env) — the ZapDB key
is ``kms/secrets/{path}/{env}/{name}`` and a write upserts it in place.
There is no version history and no versioned read, so a ``version=``
argument can only be answered with the current value. Answering it
silently would hand back something other than what was asked for; this
fails loudly instead.
"""
def secrets_url(org: str) -> str:
"""Collection URL — list (GET) and upsert (POST)."""
return f"/v1/kms/orgs/{quote(org, safe='')}/secrets"
def secret_url(org: str, path: str, name: str) -> str:
"""URL of a single secret — read (GET) and delete (DELETE).
Every segment is escaped INDIVIDUALLY. The server splits the trailing
path at its LAST slash into (path, name); escaping the joined string
would percent-encode those separators away and the server would read the
whole thing as one long name.
"""
segments = [*_path_segments(path), _checked_name(name)]
escaped = "/".join(quote(segment, safe="") for segment in segments)
return f"{secrets_url(org)}/{escaped}"
def list_params(path: str, env: str) -> dict[str, str]:
"""Query for GET /v1/kms/orgs/{org}/secrets.
``path`` may be empty here — listing is a prefix scan over
``kms/secrets/{path}/{env}/`` and an empty path is a legitimate prefix.
"""
return {"path": path, "env": _checked_env(env)}
def env_params(env: str) -> dict[str, str]:
"""Query for the single-secret read and delete routes."""
return {"env": _checked_env(env)}
def upsert_body(path: str, name: str, value: str, env: str) -> dict[str, str]:
"""Body for POST /v1/kms/orgs/{org}/secrets — create and replace alike."""
return {
"path": "/".join(_path_segments(path)),
"name": _checked_name(name),
"env": _checked_env(env),
"value": value,
}
def check_version(version: Optional[int]) -> None:
"""Reject a versioned read. See :class:`VersionUnsupportedError`."""
if version is not None:
raise VersionUnsupportedError(
f"cannot read version {version!r}: luxfi/kms holds exactly one value "
"per (path, name, env) — there is no version history to read from. "
"Drop the version argument to read the current value."
)
def names_of(data: Any) -> list[str]:
"""Decode the list response: ``{"names": [...]}``."""
if isinstance(data, dict) and isinstance(data.get("names"), list):
return [str(name) for name in data["names"]]
raise ValueError(f"unexpected KMS list response: {data!r}")
def value_of(data: Any) -> str:
"""Decode the single-secret response: ``{"secret": {"value": "..."}}``."""
if isinstance(data, dict):
secret = data.get("secret")
if isinstance(secret, dict) and "value" in secret:
return str(secret["value"])
raise ValueError(f"unexpected KMS secret response: {data!r}")
def _path_segments(path: str) -> list[str]:
segments = [segment for segment in path.split("/") if segment]
if not segments:
raise ValueError(
"path is required: the server splits the trailing URL at its last "
"slash into (path, name), so a bare name has nothing to split"
)
return segments
def _checked_name(name: str) -> str:
if not name:
raise ValueError("name is required")
if "/" in name:
raise ValueError(
f"name may not contain '/': {name!r}. The server splits the trailing "
"URL at its last slash, so a slashed name is written under one key "
"and read back under another — the write looks like it succeeded and "
"the read never finds it. Move the leading segments into path."
)
return name
def _checked_env(env: str) -> str:
if not env.strip():
raise ValueError(
f"env is required (use {DEFAULT_ENV!r} for the default bucket): env is "
"part of the storage key, and the server rejects an empty env on write"
)
return env
pkg/hanzo-kms/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "hanzo-kms"
version = "1.1.0"
version = "1.1.1"
description = "Hanzo KMS SDK - Secret management for Python"
readme = "README.md"
license = { text = "MIT" }
pkg/hanzo-kms/tests/test_clients.py
+253
View File
@@ -0,0 +1,253 @@
"""Pin what the sync and async clients actually put on the wire.
Every request is captured through an httpx MockTransport and asserted
whole — method, URL, query and body. Two invariants matter most:
* no request may touch an ``/api/`` path (the Infisical regression), and
* the sync and async clients must emit byte-identical request lines, so the
two files cannot drift apart.
"""
import asyncio
import os
from typing import Any
import httpx
import pytest
from hanzo_kms import AsyncKMSClient, ClientSettings, KMSClient, routes
from hanzo_kms.models import settings_from_env
SITE_URL = "https://kms.test"
TOKEN = "test-access-token"
VALUE = "correct horse battery staple"
SETTINGS = ClientSettings(
site_url=SITE_URL,
org="lux",
client_id="cid",
client_secret="csecret",
)
class Recorder:
"""Captures requests and answers with the server's documented shapes."""
def __init__(self) -> None:
self.requests: list[httpx.Request] = []
def __call__(self, request: httpx.Request) -> httpx.Response:
self.requests.append(request)
path = request.url.path
if path == routes.LOGIN:
return httpx.Response(200, json={"accessToken": TOKEN, "expiresIn": 3600})
if path == routes.HEALTH:
return httpx.Response(200, json={"service": "kms", "status": "ok"})
if request.method == "GET" and path.endswith("/secrets"):
return httpx.Response(200, json={"names": ["deploy-mnemonic", "bls-key"]})
if request.method == "GET":
return httpx.Response(200, json={"secret": {"value": VALUE}})
if request.method == "POST":
return httpx.Response(201, json={"ok": True})
return httpx.Response(200, json={"ok": True})
@property
def lines(self) -> list[tuple[str, str, bytes]]:
return [(r.method, str(r.url), r.content) for r in self.requests]
@property
def urls(self) -> list[str]:
return [str(r.url) for r in self.requests]
def sync_client(recorder: Recorder) -> KMSClient:
"""A client whose lazily-built transport is the recorder."""
client = KMSClient(SETTINGS.model_copy())
client._http_client = httpx.Client(
base_url=SITE_URL, transport=httpx.MockTransport(recorder)
)
return client
def async_client(recorder: Recorder) -> AsyncKMSClient:
client = AsyncKMSClient(SETTINGS.model_copy())
client._http_client = httpx.AsyncClient(
base_url=SITE_URL, transport=httpx.MockTransport(recorder)
)
return client
def record_sync() -> Recorder:
"""Drive every operation once, synchronously."""
recorder = Recorder()
with sync_client(recorder) as client:
client.health()
client.list_secrets("providers/lux", "prod")
client.get_secret("providers/lux", "deploy-mnemonic", "prod")
client.put_secret("providers/lux", "deploy-mnemonic", VALUE, "prod")
client.delete_secret("providers/lux", "deploy-mnemonic", "prod")
return recorder
def record_async() -> Recorder:
"""Drive every operation once, asynchronously — same order."""
recorder = Recorder()
async def run() -> None:
async with async_client(recorder) as client:
await client.health()
await client.list_secrets("providers/lux", "prod")
await client.get_secret("providers/lux", "deploy-mnemonic", "prod")
await client.put_secret("providers/lux", "deploy-mnemonic", VALUE, "prod")
await client.delete_secret("providers/lux", "deploy-mnemonic", "prod")
asyncio.run(run())
return recorder
@pytest.fixture(params=[record_sync, record_async], ids=["sync", "async"])
def recorded(request: Any) -> Recorder:
return request.param()
def test_no_request_touches_an_api_path(recorded: Recorder) -> None:
"""The regression pin: /api/* is Infisical's surface, never luxfi/kms's."""
assert recorded.urls
for url in recorded.urls:
assert "/api/" not in url, f"{url} regressed onto an Infisical /api/ path"
def test_full_request_sequence(recorded: Recorder) -> None:
"""Every route, exactly as cmd/kms/main.go registers it."""
assert recorded.lines == [
("GET", f"{SITE_URL}/v1/kms/healthz", b""),
("POST", f"{SITE_URL}/v1/kms/auth/login", b'{"clientId":"cid","clientSecret":"csecret"}'),
("GET", f"{SITE_URL}/v1/kms/orgs/lux/secrets?path=providers%2Flux&env=prod", b""),
("GET", f"{SITE_URL}/v1/kms/orgs/lux/secrets/providers/lux/deploy-mnemonic?env=prod", b""),
(
"POST",
f"{SITE_URL}/v1/kms/orgs/lux/secrets",
b'{"path":"providers/lux","name":"deploy-mnemonic","env":"prod","value":"%s"}'
% VALUE.encode(),
),
(
"DELETE",
f"{SITE_URL}/v1/kms/orgs/lux/secrets/providers/lux/deploy-mnemonic?env=prod",
b"",
),
]
def test_sync_and_async_are_mirror_images() -> None:
"""The two clients are one surface with two I/O models. Keep them equal."""
assert record_sync().lines == record_async().lines
def test_bearer_on_every_call_except_login_and_health(recorded: Recorder) -> None:
for request in recorded.requests:
authorization = request.headers.get("authorization")
if request.url.path in (routes.LOGIN, routes.HEALTH):
assert authorization is None
else:
assert authorization == f"Bearer {TOKEN}"
def test_org_scopes_the_url() -> None:
recorder = Recorder()
client = KMSClient(SETTINGS.model_copy(update={"org": "hanzo"}))
client._http_client = httpx.Client(
base_url=SITE_URL, transport=httpx.MockTransport(recorder)
)
client.get_secret("providers/hanzo", "deploy-mnemonic")
client.close()
assert recorder.urls[-1] == (
f"{SITE_URL}/v1/kms/orgs/hanzo/secrets/providers/hanzo/deploy-mnemonic?env=default"
)
def test_returned_values() -> None:
recorder = Recorder()
with sync_client(recorder) as client:
assert client.health() == {"service": "kms", "status": "ok"}
assert client.list_secrets("providers/lux", "prod") == ["deploy-mnemonic", "bls-key"]
assert client.get_secret("providers/lux", "deploy-mnemonic", "prod") == VALUE
def test_versioned_read_refused_before_any_request() -> None:
recorder = Recorder()
with sync_client(recorder) as client:
with pytest.raises(routes.VersionUnsupportedError):
client.get_secret("providers/lux", "deploy-mnemonic", "prod", version=2)
assert recorder.requests == []
def test_pre_issued_token_skips_login() -> None:
recorder = Recorder()
client = KMSClient(
ClientSettings(site_url=SITE_URL, org="lux", access_token="pre-issued")
)
client._http_client = httpx.Client(
base_url=SITE_URL, transport=httpx.MockTransport(recorder)
)
client.list_secrets("providers/lux", "prod")
client.close()
assert routes.LOGIN not in recorder.urls[0]
assert recorder.requests[0].headers["authorization"] == "Bearer pre-issued"
def test_missing_credentials_fail_loudly() -> None:
client = KMSClient(ClientSettings(site_url=SITE_URL, org="lux"))
with pytest.raises(ValueError, match="no KMS credentials"):
client.list_secrets("providers/lux", "prod")
def test_inject_env_lists_then_reads_each(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("deploy-mnemonic", raising=False)
monkeypatch.delenv("bls-key", raising=False)
recorder = Recorder()
with sync_client(recorder) as client:
assert client.inject_env("providers/lux", "prod") == 2
assert os.environ["deploy-mnemonic"] == VALUE
assert os.environ["bls-key"] == VALUE
# one login + one list + one read per name
assert len(recorder.requests) == 4
@pytest.mark.parametrize(
("env_var", "field", "value"),
[
("HANZO_KMS_URL", "site_url", "https://kms.lux.network"),
("HANZO_KMS_ORG", "org", "zoo"),
("HANZO_KMS_CLIENT_ID", "client_id", "an-id"),
("HANZO_KMS_CLIENT_SECRET", "client_secret", "a-secret"),
("HANZO_KMS_TOKEN", "access_token", "a-token"),
],
)
def test_settings_from_env(
monkeypatch: pytest.MonkeyPatch, env_var: str, field: str, value: str
) -> None:
monkeypatch.setenv(env_var, value)
assert getattr(settings_from_env(), field) == value
def test_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
for env_var in (
"HANZO_KMS_URL",
"HANZO_KMS_ORG",
"HANZO_KMS_CLIENT_ID",
"HANZO_KMS_CLIENT_SECRET",
"HANZO_KMS_TOKEN",
):
monkeypatch.delenv(env_var, raising=False)
settings = settings_from_env()
assert settings.site_url == "https://kms.hanzo.ai"
assert settings.org == "hanzo"
def test_both_clients_expose_the_same_methods() -> None:
"""Mirror images down to the method names."""
def public(cls: type) -> set[str]:
return {name for name in vars(cls) if not name.startswith("_")}
assert public(KMSClient) == public(AsyncKMSClient)
pkg/hanzo-kms/tests/test_routes.py
+133
View File
@@ -0,0 +1,133 @@
"""Pin the canonical luxfi/kms wire shape.
These tests are pure — no I/O — and they exist because two mistakes in this
SDK were invisible in production:
1. Infisical ``/api/*`` paths, which luxfi/kms has never served. They read as
JSON decode errors instead of 404s while old builds answered every
unmatched path with the console SPA (200 text/html).
2. Escaping the joined path instead of each segment, which percent-encodes
the separators the server splits on.
"""
from urllib.parse import unquote
import pytest
from hanzo_kms import routes
def test_no_api_paths() -> None:
"""Every route lives under /v1/kms. Nothing under /api/ — ever."""
urls = [
routes.LOGIN,
routes.HEALTH,
routes.secrets_url("hanzo"),
routes.secret_url("hanzo", "providers/lux", "deploy-mnemonic"),
]
for url in urls:
assert "/api/" not in url, f"{url} regressed onto an Infisical /api/ path"
assert url.startswith("/v1/kms/"), url
def test_secret_url_shape() -> None:
assert (
routes.secret_url("lux", "providers/lux", "deploy-mnemonic")
== "/v1/kms/orgs/lux/secrets/providers/lux/deploy-mnemonic"
)
assert routes.secrets_url("lux") == "/v1/kms/orgs/lux/secrets"
def test_each_segment_is_escaped_individually() -> None:
"""Separators survive; everything else inside a segment is encoded."""
url = routes.secret_url("my org", "a b/c+d", "e f")
assert url == "/v1/kms/orgs/my%20org/secrets/a%20b/c%2Bd/e%20f"
# The joined-string mistake would have produced this instead:
assert "a%20b%2Fc" not in url
@pytest.mark.parametrize(
("path", "name"),
[
("providers/lux", "deploy-mnemonic"),
("a", "b"),
("a/b/c/d", "e"),
("weird path/with spaces", "name with spaces"),
("q?uery#frag", "amp&sand"),
],
)
def test_server_split_round_trip(path: str, name: str) -> None:
"""Model the server: unescape the trailing path, split at the LAST slash.
cmd/kms/main.go does exactly this — ``strings.LastIndex(rest, "/")`` over
the unescaped remainder — for both GET and DELETE. Whatever this SDK
builds has to survive that split unchanged.
"""
url = routes.secret_url("hanzo", path, name)
rest = unquote(url.removeprefix("/v1/kms/orgs/hanzo/secrets/"))
got_path, _, got_name = rest.rpartition("/")
assert (got_path, got_name) == ("/".join(s for s in path.split("/") if s), name)
def test_name_with_slash_is_rejected() -> None:
"""A slashed name writes under one key and reads back under another."""
with pytest.raises(ValueError, match="may not contain"):
routes.secret_url("hanzo", "providers", "lux/deploy-mnemonic")
with pytest.raises(ValueError, match="may not contain"):
routes.upsert_body("providers", "lux/deploy-mnemonic", "v", "prod")
def test_path_is_required_for_a_single_secret() -> None:
"""No path means no slash to split on — the server 400s. Fail earlier."""
for path in ("", "/", "///"):
with pytest.raises(ValueError, match="path is required"):
routes.secret_url("hanzo", path, "name")
def test_empty_name_is_rejected() -> None:
with pytest.raises(ValueError, match="name is required"):
routes.secret_url("hanzo", "providers", "")
def test_list_path_may_be_empty() -> None:
"""Listing is a prefix scan, so an empty path is legitimate there."""
assert routes.list_params("", "prod") == {"path": "", "env": "prod"}
def test_env_defaults_to_default_and_may_not_be_blank() -> None:
assert routes.DEFAULT_ENV == "default"
assert routes.env_params(routes.DEFAULT_ENV) == {"env": "default"}
for blank in ("", " "):
with pytest.raises(ValueError, match="env is required"):
routes.env_params(blank)
with pytest.raises(ValueError, match="env is required"):
routes.upsert_body("providers", "name", "value", blank)
def test_upsert_body_shape() -> None:
assert routes.upsert_body("/providers/lux/", "deploy-mnemonic", "abc", "prod") == {
"path": "providers/lux",
"name": "deploy-mnemonic",
"env": "prod",
"value": "abc",
}
def test_version_read_is_refused() -> None:
"""KMS holds one value per (path, name, env) — a version can't be honored."""
routes.check_version(None)
with pytest.raises(routes.VersionUnsupportedError, match="one value"):
routes.check_version(3)
def test_decoders_reject_anything_but_the_documented_shape() -> None:
assert routes.names_of({"names": ["a", "b"]}) == ["a", "b"]
assert routes.names_of({"names": []}) == []
assert routes.value_of({"secret": {"value": "s3cret"}}) == "s3cret"
# An SPA catch-all or a JSON 404 must not read as an empty result.
for junk in ({"message": "not found", "path": "/api/v3/secrets/raw"}, {}, [], None):
with pytest.raises(ValueError, match="unexpected KMS"):
routes.names_of(junk)
with pytest.raises(ValueError, match="unexpected KMS"):
routes.value_of(junk)
pkg/hanzo-tools-auth/hanzo_tools/auth/session.py
+4 -22
View File
@@ -244,29 +244,11 @@ class HanzoSession:
if self._kms_client:
return self._kms_client
from hanzo_kms import KMSClient, ClientSettings
kms_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
if client_id and client_secret:
from hanzo_kms import UniversalAuthMethod, AuthenticationOptions
settings = ClientSettings(
site_url=kms_url,
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id=client_id,
client_secret=client_secret,
)
),
)
self._kms_client = KMSClient(settings=settings)
else:
# Fall back to default env-based construction
self._kms_client = KMSClient()
from hanzo_kms import KMSClient
# KMSClient() reads HANZO_KMS_URL / _ORG / _CLIENT_ID / _CLIENT_SECRET
# / _TOKEN itself — see hanzo_kms.settings_from_env.
self._kms_client = KMSClient()
return self._kms_client
def get_paas_client(self) -> Any:
pkg/hanzo-tools-kms/hanzo_tools/kms/kms_tool.py
+98 -167
View File
@@ -1,13 +1,13 @@
"""MCP tool for Hanzo KMS secret management.
Wraps the hanzo-kms client to provide secret CRUD operations
via MCP. Auth uses HANZO_KMS_CLIENT_ID / HANZO_KMS_CLIENT_SECRET
environment variables (same pattern as hanzo-cli).
Wraps the hanzo-kms client to provide secret CRUD operations via MCP. A
secret is (org, path, name, env): the org comes from HANZO_KMS_ORG, and auth
from HANZO_KMS_CLIENT_ID / HANZO_KMS_CLIENT_SECRET (or HANZO_KMS_TOKEN) —
the same pattern as hanzo-cli.
"""
from __future__ import annotations
import os
import json
import logging
from typing import Any, Annotated, final
@@ -20,45 +20,40 @@ from hanzo_tools.core.base import BaseTool
logger = logging.getLogger(__name__)
DEFAULT_ENV = "default"
DESCRIPTION = """Manage secrets via Hanzo KMS.
Requires HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET environment variables.
Requires HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET (or HANZO_KMS_TOKEN)
environment variables. HANZO_KMS_ORG selects the organization (default: hanzo).
A secret is identified by path + name + env, e.g. path="providers/lux",
name="deploy-mnemonic", env="prod".
Actions:
- list: List all secrets in a project/environment (values masked)
- get: Get a single secret value
- set: Create or update a secret
- list: List the secret names at a path
- get: Get a single secret value (masked unless reveal=true)
- set: Create or replace a secret
- delete: Remove a secret
- inject: Output secrets as export/dotenv/json format
- inject: Output every secret at a path as export/dotenv/json
"""
def _get_kms_client() -> Any:
"""Build a KMS client from environment."""
from hanzo_kms import KMSClient, ClientSettings
"""Build a KMS client from the environment.
kms_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
KMSClient() reads HANZO_KMS_URL / _ORG / _CLIENT_ID / _CLIENT_SECRET /
_TOKEN itself — see hanzo_kms.settings_from_env.
"""
from hanzo_kms import KMSClient
if client_id and client_secret:
from hanzo_kms import UniversalAuthMethod, AuthenticationOptions
settings = ClientSettings(
site_url=kms_url,
auth=AuthenticationOptions(
universal_auth=UniversalAuthMethod(
client_id=client_id,
client_secret=client_secret,
)
),
)
return KMSClient(settings=settings)
# Fall back to default env-based construction
return KMSClient()
def _masked(value: str) -> str:
return f"{value[:4]}***" if len(value) > 4 else "***"
@final
class KMSTool(BaseTool):
"""MCP tool for KMS secret management."""
@@ -75,170 +70,111 @@ class KMSTool(BaseTool):
self,
ctx: MCPContext,
action: str = "list",
project: str | None = None,
environment: str | None = None,
secret_name: str | None = None,
secret_value: str | None = None,
path: str = "/",
path: str = "",
name: str | None = None,
value: str | None = None,
env: str = DEFAULT_ENV,
format: str = "export",
reveal: bool = False,
**kwargs: Any,
) -> str:
if action == "list":
return await self._list(project, environment, path)
return await self._list(path, env)
elif action == "get":
return await self._get(project, environment, secret_name, path, reveal)
return await self._get(path, name, env, reveal)
elif action == "set":
return await self._set(project, environment, secret_name, secret_value, path)
return await self._set(path, name, value, env)
elif action == "delete":
return await self._delete(project, environment, secret_name, path)
return await self._delete(path, name, env)
elif action == "inject":
return await self._inject(project, environment, path, format)
else:
return json.dumps({"error": f"Unknown action: {action}. Use: list, get, set, delete, inject"})
async def _list(self, project: str | None, env: str | None, path: str) -> str:
if not project or not env:
return json.dumps({"error": "Required: project and environment"})
return await self._inject(path, env, format)
return json.dumps(
{"error": f"Unknown action: {action}. Use: list, get, set, delete, inject"}
)
async def _list(self, path: str, env: str) -> str:
client = _get_kms_client()
try:
secrets = client.list_secrets(project_id=project, environment=env, path=path)
names = client.list_secrets(path, env)
finally:
client.close()
if not secrets:
return json.dumps({"message": f"No secrets found in {project}/{env}", "secrets": []})
return json.dumps(
{"path": path, "env": env, "count": len(names), "names": names}, indent=2
)
rows = []
for s in secrets:
val = s.secret_value
masked = f"{val[:4]}***" if len(val) > 4 else "***"
rows.append({
"key": s.secret_key,
"value": masked,
"version": s.version,
"updated": str(s.updated_at)[:19] if s.updated_at else None,
})
return json.dumps({
"project": project,
"environment": env,
"path": path,
"count": len(rows),
"secrets": rows,
}, indent=2)
async def _get(
self, project: str | None, env: str | None, name: str | None, path: str, reveal: bool
) -> str:
if not project or not env or not name:
return json.dumps({"error": "Required: project, environment, and secret_name"})
async def _get(self, path: str, name: str | None, env: str, reveal: bool) -> str:
if not name:
return json.dumps({"error": "Required: name"})
client = _get_kms_client()
try:
secret = client.get_secret(
project_id=project, environment=env, secret_name=name, path=path
)
value = client.get_secret(path, name, env)
finally:
client.close()
val = secret.secret_value
if not reveal:
val = f"{val[:4]}***" if len(val) > 4 else "***"
return json.dumps(
{
"path": path,
"name": name,
"env": env,
"value": value if reveal else _masked(value),
"revealed": reveal,
},
indent=2,
)
return json.dumps({
"key": secret.secret_key,
"value": val,
"version": secret.version,
"type": secret.type,
"environment": secret.environment,
"comment": secret.secret_comment or None,
"revealed": reveal,
}, indent=2)
async def _set(
self, project: str | None, env: str | None, name: str | None, value: str | None, path: str
) -> str:
if not project or not env or not name or not value:
return json.dumps({"error": "Required: project, environment, secret_name, and secret_value"})
async def _set(self, path: str, name: str | None, value: str | None, env: str) -> str:
if not name or value is None:
return json.dumps({"error": "Required: name and value"})
client = _get_kms_client()
try:
# Try update first, create if it doesn't exist
try:
secret = client.update_secret(
project_id=project,
environment=env,
secret_name=name,
secret_value=value,
)
return json.dumps({
"action": "updated",
"key": name,
"version": secret.version,
})
except Exception:
secret = client.create_secret(
project_id=project,
environment=env,
secret_name=name,
secret_value=value,
)
return json.dumps({
"action": "created",
"key": name,
})
# One upsert — KMS holds exactly one value per (path, name, env).
client.put_secret(path, name, value, env)
finally:
client.close()
async def _delete(
self, project: str | None, env: str | None, name: str | None, path: str
) -> str:
if not project or not env or not name:
return json.dumps({"error": "Required: project, environment, and secret_name"})
return json.dumps({"action": "set", "path": path, "name": name, "env": env})
async def _delete(self, path: str, name: str | None, env: str) -> str:
if not name:
return json.dumps({"error": "Required: name"})
client = _get_kms_client()
try:
client.delete_secret(
project_id=project,
environment=env,
secret_name=name,
path=path,
)
client.delete_secret(path, name, env)
finally:
client.close()
return json.dumps({"action": "deleted", "key": name})
async def _inject(self, project: str | None, env: str | None, path: str, fmt: str) -> str:
if not project or not env:
return json.dumps({"error": "Required: project and environment"})
return json.dumps({"action": "deleted", "path": path, "name": name, "env": env})
async def _inject(self, path: str, env: str, fmt: str) -> str:
client = _get_kms_client()
try:
secrets = client.list_secrets(project_id=project, environment=env, path=path)
# One list plus one read per name — the list route returns names.
values = {
name: client.get_secret(path, name, env)
for name in client.list_secrets(path, env)
}
finally:
client.close()
if not secrets:
if not values:
return json.dumps({"message": "No secrets found", "output": ""})
if fmt == "json":
data = {s.secret_key: s.secret_value for s in secrets}
return json.dumps(data, indent=2)
elif fmt == "dotenv":
lines = []
for s in secrets:
val = s.secret_value.replace('"', '\\"')
lines.append(f'{s.secret_key}="{val}"')
return "\n".join(lines)
else: # export
lines = []
for s in secrets:
val = s.secret_value.replace("'", "'\\''")
lines.append(f"export {s.secret_key}='{val}'")
return "\n".join(lines)
return json.dumps(values, indent=2)
lines = []
for name, value in values.items():
if fmt == "dotenv":
dotenv_safe = value.replace('"', '\\"')
lines.append(f'{name}="{dotenv_safe}"')
else: # export
shell_safe = value.replace("'", "'\\''")
lines.append(f"export {name}='{shell_safe}'")
return "\n".join(lines)
def register(self, mcp_server: FastMCP) -> None:
"""Register KMS tool with explicit parameters."""
@@ -253,26 +189,22 @@ class KMSTool(BaseTool):
str,
Field(description="Action: list, get, set, delete, inject"),
] = "list",
project: Annotated[
path: Annotated[
str,
Field(description="Secret path, e.g. providers/lux"),
] = "",
name: Annotated[
str | None,
Field(description="KMS project ID or slug"),
Field(description="Secret name (for get/set/delete). No slashes."),
] = None,
environment: Annotated[
str | None,
Field(description="Environment: dev, staging, production"),
] = None,
secret_name: Annotated[
str | None,
Field(description="Secret key name (for get/set/delete)"),
] = None,
secret_value: Annotated[
value: Annotated[
str | None,
Field(description="Secret value (for set)"),
] = None,
path: Annotated[
env: Annotated[
str,
Field(description="Secret path prefix (default: /)"),
] = "/",
Field(description="Environment bucket, e.g. dev, prod"),
] = DEFAULT_ENV,
format: Annotated[
str,
Field(description="Output format for inject: export, dotenv, json"),
@@ -286,11 +218,10 @@ class KMSTool(BaseTool):
return await tool_instance.call(
ctx,
action=action,
project=project,
environment=environment,
secret_name=secret_name,
secret_value=secret_value,
path=path,
name=name,
value=value,
env=env,
format=format,
reveal=reveal,
)