Build an AI Kubernetes Secret Rotation Assistant with Claude API

pip install anthropic kubernetes

import anthropic
from kubernetes import client, config
 
client_ai = anthropic.Anthropic()
 
 
def find_secret_consumers(secret_name: str, namespace: str) -> dict:
    """Find every workload that references this secret — as an env var,
    a mounted volume, or an imagePullSecret."""
    config.load_kube_config()
    v1 = client.CoreV1Api()
    apps_v1 = client.AppsV1Api()
 
    consumers = {"deployments": [], "statefulsets": [], "cronjobs": [], "service_accounts": []}
 
    for deploy in apps_v1.list_namespaced_deployment(namespace).items:
        if references_secret(deploy.spec.template.spec, secret_name):
            consumers["deployments"].append(deploy.metadata.name)
 
    for sts in apps_v1.list_namespaced_stateful_set(namespace).items:
        if references_secret(sts.spec.template.spec, secret_name):
            consumers["statefulsets"].append(sts.metadata.name)
 
    for sa in v1.list_namespaced_service_account(namespace).items:
        if sa.secrets and any(s.name == secret_name for s in sa.secrets):
            consumers["service_accounts"].append(sa.metadata.name)
 
    return consumers
 
 
def references_secret(pod_spec, secret_name: str) -> bool:
    for container in pod_spec.containers:
        for env in (container.env or []):
            if env.value_from and env.value_from.secret_key_ref and env.value_from.secret_key_ref.name == secret_name:
                return True
    for volume in (pod_spec.volumes or []):
        if volume.secret and volume.secret.secret_name == secret_name:
            return True
    return False

def check_external_references(secret_name: str, namespace: str) -> list[str]:
    """Secrets referenced from outside the cluster (CI systems pulling
    from Vault-synced secrets, external services reading via API) are
    the ones that break rotation silently — check what you can."""
    warnings = []
 
    # Check for external-secrets operator syncing this from a source
    result = subprocess.run(
        ["kubectl", "get", "externalsecret", "-n", namespace, "-o", "json"],
        capture_output=True, text=True
    )
    import json
    for es in json.loads(result.stdout).get("items", []):
        if es["spec"]["target"]["name"] == secret_name:
            warnings.append(f"Synced by ExternalSecret from {es['spec']['secretStoreRef']['name']} — rotation must originate at the source, not the K8s secret")
 
    return warnings

PLAN_PROMPT = """Generate a safe rotation plan for this Kubernetes secret.
 
Secret: {secret_name} in namespace {namespace}
Known in-cluster consumers: {consumers}
External sync warnings: {external_warnings}
 
Generate a step-by-step rotation plan that avoids downtime:
1. Whether this credential type supports dual validity (old and new both
   work simultaneously during a grace period) — most API keys and OAuth
   secrets do, most database passwords do NOT unless you create a second
   user first
2. The safe order of operations: create new credential -> update secret
   with BOTH old and new (if dual-validity) or a rolling restart sequence
   (if not) -> verify consumers healthy -> revoke old credential
3. Specific kubectl commands for restarting each consumer type to pick
   up the new secret (Deployments don't auto-restart on Secret change
   unless using something like Reloader)
4. Rollback plan if a consumer fails to authenticate with the new credential
 
Respond as a step-by-step runbook."""
 
 
def generate_rotation_plan(secret_name: str, namespace: str, consumers: dict, external_warnings: list) -> str:
    response = client_ai.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": PLAN_PROMPT.format(
                secret_name=secret_name, namespace=namespace,
                consumers=consumers, external_warnings=external_warnings,
            )
        }]
    )
    return response.content[0].text