znvault 1.2.2

Python SDK for ZN-Vault secrets management

ZN-Vault Python SDK

A Python client library for ZN-Vault secrets management system.

PyPI: https://pypi.org/project/znvault/

Installation

pip install znvault

Or with a specific version:

pip install znvault==1.0.0

Or install from source:

git clone https://github.com/zincware/zn-vault-sdk-python.git
cd zn-vault-sdk-python
pip install -e .

Quick Start

from znvault import ZnVaultClient, SecretType, CreateSecretRequest

# Create client with API key
client = ZnVaultClient.create(
    "https://vault.example.com:8443",
    api_key="znv_xxxx"
)

# Or use the builder pattern
client = (
    ZnVaultClient.builder()
    .base_url("https://vault.example.com:8443")
    .api_key("znv_xxxx")
    .timeout(60)
    .trust_self_signed(True)  # For development
    .build()
)

# Check health
health = client.health.check()
print(f"Status: {health.status}")

# Login with username/password
auth = client.auth.login("username", "password")
print(f"Logged in: {auth.access_token[:30]}...")

Authentication

Username/Password

# Login
result = client.auth.login("alice", "password123", totp_code="123456")

# Refresh token
new_tokens = client.auth.refresh()

# Get current user
user = client.auth.me()
print(f"Username: {user.username}")

# Logout
client.auth.logout()

API Keys

# Create API key
key = client.auth.create_api_key("my-service", expires_in="90d")
print(f"Key: {key.key}")  # Only shown once

# List API keys
keys = client.auth.list_api_keys()

# Revoke API key
client.auth.revoke_api_key(key.id)

Secrets Management

Create Secrets

from znvault import CreateSecretRequest, SecretType

# Create a credential secret
request = CreateSecretRequest(
    alias="api/production/db-creds",
    tenant="acme",
    type=SecretType.CREDENTIAL,
    data={"username": "dbuser", "password": "secret123"},
    tags=["production", "database"]
)
secret = client.secrets.create(request)
print(f"Created: {secret.id}")

Retrieve Secrets

# Get metadata by ID
secret = client.secrets.get("secret-id")

# Get by tenant and alias
secret = client.secrets.get_by_alias("acme", "api/production/db-creds")

# Decrypt secret value
data = client.secrets.decrypt("secret-id")
password = data.data["password"]

Update and Delete

from znvault import UpdateSecretRequest

# Update secret (creates new version)
update = UpdateSecretRequest(
    data={"username": "newuser", "password": "newpass"}
)
secret = client.secrets.update("secret-id", update)
print(f"New version: {secret.version}")

# Delete secret
client.secrets.delete("secret-id")

List and Filter

from znvault import SecretFilter, SecretType

# List with filters
filter = SecretFilter(
    tenant="acme",
    env="production",
    type=SecretType.CREDENTIAL,
    limit=100
)
secrets = client.secrets.list(filter)

File Upload/Download

# Upload a file as a secret
secret = client.secrets.upload_file(
    alias="ssl/production/cert",
    tenant="acme",
    file_path="/path/to/cert.pem",
    tags=["certificate", "ssl"]
)

# Download a file secret
client.secrets.download_file("secret-id", "/path/to/output.pem")

KMS Operations

Key Management

from znvault import CreateKeyRequest, KeySpec, KeyUsage

# Create a KMS key
request = CreateKeyRequest(
    alias="alias/my-encryption-key",
    tenant="acme",
    description="Production encryption key",
    key_spec=KeySpec.AES_256,
    usage=KeyUsage.ENCRYPT_DECRYPT,
    rotation_enabled=True,
    rotation_days=90
)
key = client.kms.create_key(request)
print(f"Key ID: {key.key_id}")

# List keys
keys = client.kms.list_keys()

Encrypt/Decrypt

import base64

# Encrypt data
plaintext = b"sensitive data"
result = client.kms.encrypt("key-id", plaintext)
print(f"Ciphertext: {result.ciphertext}")

# Decrypt data
decrypted = client.kms.decrypt_bytes("key-id", result.ciphertext)
print(f"Decrypted: {decrypted.decode()}")

Data Keys

# Generate data key for envelope encryption
data_key = client.kms.generate_data_key("key-id")
# Use data_key.plaintext to encrypt locally
# Store data_key.ciphertext with the encrypted data

Admin Operations

Tenants

from znvault import CreateTenantRequest

# Create tenant
request = CreateTenantRequest(name="newcorp", display_name="New Corp Inc")
tenant = client.tenants.create(request)

# List tenants
tenants = client.tenants.list()

Users

from znvault import CreateUserRequest

# Create user
request = CreateUserRequest(
    username="bob",
    password="secure123",
    email="[e-mail]",
    role="admin",
    tenant_id="acme"
)
user = client.users.create(request)

# List users
users = client.users.list(tenant_id="acme")

Roles

from znvault import CreateRoleRequest

# Create role
request = CreateRoleRequest(
    name="SecretReader",
    description="Can read secrets",
    permissions=["secret:read:*"]
)
role = client.roles.create(request)

# List roles
roles = client.roles.list(include_system=True)

Policies

from znvault import PolicyDocument, PolicyStatement, PolicyEffect

# Create ABAC policy
document = PolicyDocument(
    statements=[
        PolicyStatement(
            effect=PolicyEffect.ALLOW,
            actions=["secret:read:*"],
            resources=["secret:acme/*"]
        )
    ]
)

policy = client.policies.create(
    name="acme-secret-reader",
    document=document,
    tenant_id="acme"
)

Audit Logs

from znvault import AuditFilter
from datetime import datetime, timedelta

# List audit entries
filter = AuditFilter(
    action="secret:read",
    start_date=datetime.now() - timedelta(days=7),
    limit=100
)
entries = client.audit.list(filter)

# Verify audit chain integrity
result = client.audit.verify()
print(f"Chain valid: {result.valid}")

Error Handling

from znvault import (
    ZnVaultError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

try:
    secret = client.secrets.decrypt("invalid-id")
except NotFoundError as e:
    print(f"Secret not found: {e.resource_id}")
except AuthorizationError as e:
    print(f"Access denied: {e.message}")
except RateLimitError as e:
    print(f"Rate limited, retry after: {e.retry_after}s")
except ZnVaultError as e:
    print(f"Error [{e.status_code}]: {e.message}")

Development

Run Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run unit tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=znvault

# Run integration tests (requires running vault)
./test-integration.sh

Type Checking

mypy src/znvault

Linting

ruff check src/

License

Apache-2.0