Secrets Service

from kubiya import ControlPlaneClient

# Initialize the client
client = ControlPlaneClient(api_key="your-api-key")

# List all secrets (metadata only)
secrets = client.secrets.list()
for secret in secrets:
    print(f"Secret: {secret['name']} - Type: {secret['type']}")

# Get specific secret value (use with caution)
secret_value = client.secrets.get_value(name="github-api-token")
print(f"Token: {secret_value['value']}")  # Handle securely!

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# List all accessible secrets (metadata only)
secrets = client.secrets.list()

print(f"Total secrets: {len(secrets)}")
for secret in secrets:
    print(f"Name: {secret['name']}")
    print(f"Type: {secret['type']}")
    print(f"Scope: {secret['scope']}")
    print(f"Created: {secret['created_at']}")
    print(f"Description: {secret.get('description', 'N/A')}")
    print("---")

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Retrieve actual secret value
secret = client.secrets.get_value(name="github-api-token")

# Access the secret value
token = secret['value']

# Use the secret (example: API call)
import requests
headers = {"Authorization": f"Bearer {token}"}
response = requests.get("https://api.github.com/user", headers=headers)

# Clear sensitive data from memory after use
token = None
secret = None

from kubiya import ControlPlaneClient
from collections import defaultdict

def generate_secret_inventory(client: ControlPlaneClient):
    """Generate comprehensive secret inventory."""
    secrets = client.secrets.list()

    # Categorize by type
    by_type = defaultdict(list)
    by_scope = defaultdict(list)
    by_integration = defaultdict(list)

    for secret in secrets:
        by_type[secret['type']].append(secret)
        by_scope[secret['scope']].append(secret)
        if secret.get('integration'):
            by_integration[secret['integration']].append(secret)

    print("Secret Inventory Report")
    print("=" * 60)
    print(f"Total Secrets: {len(secrets)}\n")

    print("By Type:")
    for secret_type, items in by_type.items():
        print(f"  {secret_type}: {len(items)}")

    print("\nBy Scope:")
    for scope, items in by_scope.items():
        print(f"  {scope}: {len(items)}")

    print("\nBy Integration:")
    for integration, items in by_integration.items():
        print(f"  {integration}: {len(items)}")

    return {
        "total": len(secrets),
        "by_type": dict(by_type),
        "by_scope": dict(by_scope),
        "by_integration": dict(by_integration)
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
inventory = generate_secret_inventory(client)

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import ControlPlaneError
from typing import List

def validate_required_secrets(
    client: ControlPlaneClient,
    required_secrets: List[str]
) -> dict:
    """Validate that all required secrets are available."""
    try:
        # Get all available secrets
        available_secrets = client.secrets.list()
        available_names = {s['name'] for s in available_secrets}

        # Check for missing secrets
        missing = []
        present = []

        for secret_name in required_secrets:
            if secret_name in available_names:
                present.append(secret_name)
            else:
                missing.append(secret_name)

        # Report results
        if missing:
            print(f"❌ Missing {len(missing)} required secrets:")
            for name in missing:
                print(f"  - {name}")
        else:
            print(f"✅ All {len(required_secrets)} required secrets are available")

        return {
            "valid": len(missing) == 0,
            "present": present,
            "missing": missing,
            "total_required": len(required_secrets)
        }

    except ControlPlaneError as e:
        print(f"Failed to validate secrets: {e}")
        return None

# Usage
client = ControlPlaneClient(api_key="your-api-key")

required = [
    "github-api-token",
    "aws-credentials",
    "slack-webhook-url",
    "datadog-api-key"
]

validation = validate_required_secrets(client, required)

if not validation['valid']:
    print(f"\nPlease create the missing secrets in Kubiya Dashboard")

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import ControlPlaneError
from contextlib import contextmanager
import gc

@contextmanager
def secure_secret(client: ControlPlaneClient, secret_name: str):
    """Context manager for secure secret access with automatic cleanup."""
    secret_value = None
    try:
        # Retrieve secret
        secret = client.secrets.get_value(name=secret_name)
        secret_value = secret['value']
        yield secret_value

    except ControlPlaneError as e:
        print(f"Failed to access secret '{secret_name}': {e}")
        raise

    finally:
        # Clear secret from memory
        if secret_value:
            secret_value = None
        gc.collect()  # Force garbage collection

# Usage
client = ControlPlaneClient(api_key="your-api-key")

with secure_secret(client, "github-api-token") as token:
    # Use the token
    import requests
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get("https://api.github.com/user", headers=headers)
    print(f"GitHub user: {response.json()['login']}")

# Token is automatically cleared from memory here

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta

def check_secret_rotation_needed(client: ControlPlaneClient, days_threshold: int = 90):
    """Identify secrets that may need rotation based on age."""
    secrets = client.secrets.list()
    now = datetime.utcnow()

    needs_rotation = []
    recently_rotated = []

    for secret in secrets:
        created_at = datetime.fromisoformat(secret['created_at'].replace('Z', '+00:00'))
        age_days = (now - created_at).days

        if age_days > days_threshold:
            needs_rotation.append({
                "name": secret['name'],
                "age_days": age_days,
                "created_at": secret['created_at'],
                "type": secret['type']
            })
        elif age_days < 30:
            recently_rotated.append(secret['name'])

    print(f"Secret Rotation Analysis (threshold: {days_threshold} days)")
    print("=" * 60)

    if needs_rotation:
        print(f"\n⚠️  {len(needs_rotation)} secrets may need rotation:")
        for item in sorted(needs_rotation, key=lambda x: x['age_days'], reverse=True):
            print(f"  - {item['name']}: {item['age_days']} days old ({item['type']})")
    else:
        print("\n✅ No secrets exceed rotation threshold")

    if recently_rotated:
        print(f"\n✅ {len(recently_rotated)} secrets rotated recently:")
        for name in recently_rotated:
            print(f"  - {name}")

    return {
        "needs_rotation": needs_rotation,
        "recently_rotated": recently_rotated,
        "threshold_days": days_threshold
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
rotation_report = check_secret_rotation_needed(client, days_threshold=90)

from kubiya import ControlPlaneClient
from typing import Dict, List

def check_integration_secrets(
    client: ControlPlaneClient,
    integration_requirements: Dict[str, List[str]]
) -> dict:
    """Check if required secrets exist for each integration."""
    secrets = client.secrets.list()

    # Build secret name to integration mapping
    available_by_integration = {}
    for secret in secrets:
        integration = secret.get('integration')
        if integration:
            if integration not in available_by_integration:
                available_by_integration[integration] = []
            available_by_integration[integration].append(secret['name'])

    # Check requirements
    results = {}
    for integration, required_secrets in integration_requirements.items():
        available = available_by_integration.get(integration, [])
        missing = [s for s in required_secrets if s not in available]

        results[integration] = {
            "required": required_secrets,
            "available": available,
            "missing": missing,
            "complete": len(missing) == 0
        }

    # Print report
    print("Integration Secrets Report")
    print("=" * 60)

    for integration, result in results.items():
        status = "✅" if result['complete'] else "❌"
        print(f"\n{status} {integration}:")
        print(f"  Required: {len(result['required'])}")
        print(f"  Available: {len(result['available'])}")

        if result['missing']:
            print(f"  Missing: {', '.join(result['missing'])}")

    return results

# Usage
client = ControlPlaneClient(api_key="your-api-key")

requirements = {
    "github": ["github-api-token", "github-webhook-secret"],
    "aws": ["aws-credentials", "aws-session-token"],
    "slack": ["slack-webhook-url", "slack-bot-token"],
    "datadog": ["datadog-api-key", "datadog-app-key"]
}

integration_report = check_integration_secrets(client, requirements)

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import ControlPlaneError
from kubiya.core.exceptions import (
    APIError as KubiyaAPIError,
    AuthenticationError as KubiyaAuthenticationError,
    TimeoutError as KubiyaTimeoutError
)

client = ControlPlaneClient(api_key="your-api-key")

# Handle secret access errors
try:
    secrets = client.secrets.list()
except ControlPlaneError as e:
    print(f"Failed to list secrets: {e}")
except KubiyaAuthenticationError as e:
    print(f"Authentication failed: {e}")
except KubiyaAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")

# Handle secret not found
try:
    secret = client.secrets.get_value(name="non-existent-secret")
except ControlPlaneError as e:
    if "not found" in str(e).lower():
        print("Secret not found")
    elif "permission denied" in str(e).lower():
        print("Insufficient permissions to access secret")
    else:
        print(f"Error accessing secret: {e}")

# Handle permission errors
try:
    secret = client.secrets.get_value(name="admin-only-secret")
except ControlPlaneError as e:
    print(f"Access denied: {e}")
    # Log audit trail or alert security team

from kubiya import ControlPlaneClient
import logging

client = ControlPlaneClient(api_key="your-api-key")

# ❌ BAD - Never do this
secret = client.secrets.get_value(name="api-token")
logging.info(f"Token: {secret['value']}")  # NEVER LOG SECRET VALUES!

# ✅ GOOD - Log metadata only
secrets = client.secrets.list()
logging.info(f"Found {len(secrets)} secrets")
logging.info(f"Secret names: {[s['name'] for s in secrets]}")

from contextlib import contextmanager
from kubiya import ControlPlaneClient

@contextmanager
def get_secret_safely(client: ControlPlaneClient, name: str):
    """Safely retrieve and cleanup secrets."""
    secret_value = None
    try:
        secret = client.secrets.get_value(name=name)
        yield secret['value']
    finally:
        secret_value = None  # Clear from memory

# Usage
client = ControlPlaneClient(api_key="your-api-key")
with get_secret_safely(client, "api-token") as token:
    # Use token here
    pass
# Token is cleared automatically

from kubiya import ControlPlaneClient

def ensure_secrets_available(client: ControlPlaneClient, required: list) -> bool:
    """Check secrets exist before attempting to use them."""
    available = {s['name'] for s in client.secrets.list()}
    missing = [name for name in required if name not in available]

    if missing:
        raise ValueError(f"Missing required secrets: {', '.join(missing)}")

    return True

# Usage
client = ControlPlaneClient(api_key="your-api-key")
ensure_secrets_available(client, ["github-token", "aws-key"])

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta
from typing import Optional

class SecretCache:
    """In-memory secret cache with TTL (use with caution!)."""

    def __init__(self, client: ControlPlaneClient, ttl_seconds: int = 300):
        self.client = client
        self.ttl_seconds = ttl_seconds
        self._cache = {}

    def get(self, name: str) -> str:
        """Get secret with caching."""
        now = datetime.utcnow()

        # Check cache
        if name in self._cache:
            value, expiry = self._cache[name]
            if now < expiry:
                return value

        # Fetch and cache
        secret = self.client.secrets.get_value(name=name)
        value = secret['value']
        expiry = now + timedelta(seconds=self.ttl_seconds)
        self._cache[name] = (value, expiry)

        return value

    def clear(self):
        """Clear all cached secrets."""
        self._cache = {}

# Usage (use only when absolutely necessary)
cache = SecretCache(client, ttl_seconds=300)
token = cache.get("api-token")

{
    "name": "github-api-token",
    "type": "token",  # token, credentials, url, certificate, env_var
    "scope": "organization",  # organization, team, user
    "description": "GitHub API token for repository access",
    "created_at": "2025-11-15T10:00:00Z",
    "created_by": "[email protected]",
    "last_used": "2025-12-11T09:30:00Z",
    "integration": "github"  # Optional: associated integration
}

{
    "name": "github-api-token",
    "value": "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",  # Actual secret value
    "type": "token",
    "scope": "organization",
    "retrieved_at": "2025-12-11T10:30:00Z"
}