​
Secrets Service

The Kubiya Secrets service provides a secure interface for managing sensitive data and credentials through the Kubiya platform. It enables you to create, retrieve, update, and delete secrets with robust validation and security features.

​
Features

  • Secure Secret Management: Create and manage encrypted secrets
  • Flexible Input Options: Support for both direct value input and file-based secrets
  • Value Retrieval: Securely retrieve secret values when needed
  • Comprehensive CRUD Operations: Full create, read, update, delete functionality
  • Validation: Built-in validation for secret names and values
  • Error Handling: Detailed error reporting with security-aware messaging

​
Core Components

​
SecretService

The main service class provides comprehensive secret management operations: 
from kubiya_workflow_sdk import KubiyaClient

# Initialize client
client = KubiyaClient(
    api_key="your-api-key",
    base_url="https://api.kubiya.ai"
)

# Access secrets service
secrets = client.secrets

​
Quick Start

​
Basic Usage

from kubiya_workflow_sdk import KubiyaClient
from kubiya_workflow_sdk.kubiya_services.exceptions import SecretError, SecretValidationError

# Initialize client
client = KubiyaClient(
    api_key="your-api-key",
    base_url="https://api.kubiya.ai"
)

try:
    # Create a new secret
    create_result = client.secrets.create(
        name="database-password",
        value="super-secure-password-123",
        description="Main database connection password"
    )
    print(f"Secret created: {create_result}")
    
    # List all secrets
    secrets_list = client.secrets.list()
    print(f"Available secrets: {secrets_list}")
    
    # Get secret details (metadata only)
    secret_info = client.secrets.get("database-password")
    print(f"Secret info: {secret_info}")
    
    # Get the actual secret value
    secret_value = client.secrets.value("database-password")
    print(f"Secret value: {'*' * len(secret_value)}")  # Don't log actual values!
    
    # Update the secret
    update_result = client.secrets.update(
        name="database-password",
        value="new-super-secure-password-456",
        description="Updated database connection password"
    )
    print(f"Secret updated: {update_result}")
    
    # Delete the secret
    delete_result = client.secrets.delete("database-password")
    print(f"Secret deleted: {delete_result}")
            
except SecretValidationError as e:
    print(f"Validation error: {e}")
    
except SecretError as e:
    print(f"Secret operation failed: {e}")

​
File-Based Secrets

For larger secrets or when reading from configuration files: 
import tempfile
from pathlib import Path

# Create a temporary file with secret content
temp_file = Path(tempfile.mktemp())
temp_file.write_text("my-secret-api-key-from-file")

try:
    # Create secret from file
    create_result = client.secrets.create(
        name="api-key-from-file",
        description="API key loaded from file",
        from_file=str(temp_file)
    )
    print(f"Secret created from file: {create_result}")
    
    # Update secret from another file
    temp_file.write_text("updated-api-key-from-file")
    update_result = client.secrets.update(
        name="api-key-from-file",
        description="Updated API key loaded from file",
        from_file=str(temp_file)
    )
    print(f"Secret updated from file: {update_result}")
    
finally:
    # Clean up
    if temp_file.exists():
        temp_file.unlink()

​
Working with Configuration Files

# Read secrets from environment or config files
import os
from pathlib import Path

config_dir = Path("./config")
if config_dir.exists():
    # Load database credentials
    db_password_file = config_dir / "db_password.txt"
    if db_password_file.exists():
        client.secrets.create(
            name="prod-db-password",
            description="Production database password",
            from_file=str(db_password_file)
        )
    
    # Load API keys
    api_key_file = config_dir / "api_key.txt"
    if api_key_file.exists():
        client.secrets.create(
            name="external-api-key",
            description="External service API key",
            from_file=str(api_key_file)
        )

​
Error Handling

The Secrets service provides specialized exceptions for different failure scenarios:

​
SecretValidationError

Thrown when secret validation fails: 
try:
    # This will fail - empty secret name
    client.secrets.create(name="", value="some-value")
except SecretValidationError as e:
    print(f"Validation failed: {e}")
    # Handle validation errors - usually user input issues

try:
    # This will fail - no value provided
    client.secrets.create(name="my-secret")
except SecretValidationError as e:
    print(f"Validation failed: {e}")
    # Secret value must be provided via value or from_file

try:
    # This will fail - both value and from_file provided
    client.secrets.create(
        name="my-secret", 
        value="direct-value",
        from_file="/path/to/file"
    )
except SecretValidationError as e:
    print(f"Validation failed: {e}")
    # Cannot use both value and from_file

​
SecretError

Thrown when secret operations fail: 
try:
    # This will fail if file doesn't exist
    client.secrets.create(
        name="my-secret",
        from_file="/nonexistent/file.txt"
    )
except SecretError as e:
    print(f"Secret operation failed: {e}")
    print(f"Secret name: {e.details.get('secret_name') if e.details else 'Unknown'}")
    
    # Handle file not found or other operational errors
    if "File not found" in str(e):
        print("Check that the specified file exists and is readable")

try:
    # This will fail if secret doesn't exist
    secret_value = client.secrets.value("nonexistent-secret")
except SecretError as e:
    print(f"Secret retrieval failed: {e}")
    # Handle secret not found errors

​
Best Practices

​
1. Never Log Secret Values

# Good practice: Don't log actual secret values
secret_value = client.secrets.value("my-secret")
print(f"Retrieved secret of length: {len(secret_value)}")

# Bad practice: Don't do this!
# print(f"Secret value: {secret_value}")  # Never log secrets!

​
2. Use Descriptive Names and Descriptions

# Good practice: Use clear, descriptive names
client.secrets.create(
    name="prod-database-master-password",
    value="secure-password",
    description="Master password for production PostgreSQL database cluster"
)

# Good practice: Environment-specific naming
environment = "staging"
client.secrets.create(
    name=f"{environment}-api-gateway-key",
    value="api-key-value",
    description=f"API gateway authentication key for {environment} environment"
)

​
3. Handle File Operations Safely

from pathlib import Path

def create_secret_from_file_safely(name: str, file_path: str, description: str = None):
    """Safely create a secret from a file with proper error handling"""
    path = Path(file_path)
    
    if not path.exists():
        raise FileNotFoundError(f"Secret file not found: {file_path}")
    
    if not path.is_file():
        raise ValueError(f"Path is not a file: {file_path}")
    
    try:
        return client.secrets.create(
            name=name,
            description=description,
            from_file=str(path)
        )
    except SecretError as e:
        print(f"Failed to create secret from file: {e}")
        raise

​
4. Implement Proper Secret Rotation

def rotate_secret(secret_name: str, new_value: str):
    """Safely rotate a secret value"""
    try:
        # Verify secret exists before rotation
        current_info = client.secrets.get(secret_name)
        print(f"Rotating secret: {current_info.get('name', secret_name)}")
        
        # Update with new value
        result = client.secrets.update(
            name=secret_name,
            value=new_value,
            description=f"Rotated on {datetime.now().isoformat()}"
        )
        
        print(f"Secret rotation successful: {result}")
        return result
        
    except SecretError as e:
        print(f"Secret rotation failed: {e}")
        raise

​
5. Validate Secret Access Patterns

def get_secret_safely(secret_name: str) -> str:
    """Get a secret value with proper error handling"""
    try:
        # Validate the secret name format
        if not secret_name or not isinstance(secret_name, str):
            raise SecretValidationError("Secret name must be a non-empty string")
        
        # Retrieve the secret value
        value = client.secrets.value(secret_name)
        
        if not value:
            raise SecretError(f"Secret '{secret_name}' returned empty value")
        
        return value
        
    except SecretValidationError as e:
        print(f"Invalid secret name: {e}")
        raise
    except SecretError as e:
        print(f"Failed to retrieve secret '{secret_name}': {e}")
        raise

​
Security Considerations

​
1. Secret Naming Conventions

  • Use descriptive but not revealing names
  • Include environment prefixes when applicable
  • Avoid including sensitive information in names

​
2. Access Patterns

  • Retrieve secrets only when needed
  • Don’t cache secret values in memory longer than necessary
  • Use proper error handling to avoid exposing sensitive information

​
3. File-Based Secrets

  • Ensure files have proper permissions (600 or 400)
  • Remove temporary files after use
  • Validate file paths to prevent directory traversal

​
Integration Examples

The Secrets service integrates seamlessly with other Kubiya services and workflows: 
# Use with workflows
from kubiya_workflow_sdk.dsl import workflow, step

@workflow
def deploy_with_secrets():
    """Example workflow using secrets"""
    
    @step(name="get-database-credentials")
    def get_db_creds():
        db_password = client.secrets.value("prod-db-password")
        api_key = client.secrets.value("external-api-key")
        return {"db_password": db_password, "api_key": api_key}
    
    @step(name="deploy-application")
    def deploy_app(credentials):
        # Use credentials for deployment
        pass

​
Next Steps