export_cluster.py
#!/usr/bin/env python3
"""Exporta el estado declarativo del clúster a cluster/, sin secretos.
Genera:
cluster/helm/releases.yaml índice de releases de Helm (chart y versión)
cluster/helm/<ns>/<release>.yaml values aplicados por el usuario (helm get values)
cluster/manifests/<ns>/<kind>-<name>.yaml
recursos aplicados a mano (kubectl apply), sin
los gestionados por Helm, ArgoCD, operadores o k3s
cluster/secrets-inventory.yaml qué Secrets existen y qué claves tienen (sin valores)
Cualquier valor que parezca una credencial se sustituye por <REDACTED> antes de
escribirse al disco. Al final se hace un escaneo de seguridad y, si encuentra algo
que parezca una clave privada o un token, el script falla.
Uso (desde la raíz del repo, con kubectl y helm apuntando al clúster):
python3 scripts/export_cluster.py
"""
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parent.parent
OUT = ROOT / "cluster"
REDACTED = "<REDACTED>"
# Claves cuyo valor es una credencial.
SECRET_KEY_RE = re.compile(
r"(pass(word|wd)?|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|"
r"credential|auth[_-]?pass|license|webhook[_-]?url|dsn)$",
re.I,
)
# Claves que contienen el NOMBRE de un Secret o de una clave, no su valor: se conservan.
SAFE_KEY_RE = re.compile(
r"(existing|secretName|secretRef|keyRef|SecretKey|Key$|Name$|File$|Path$|"
r"Ref$|Enabled$|^create|^use|Annotations$|Labels$)",
re.I,
)
# Credenciales embebidas en textos (ficheros de configuración dentro de ConfigMaps...).
INLINE_RE = re.compile(
r"(?im)^(\s*[\w.-]*(pass(word)?|secret|token|api[_-]?key|auth_pass)[\w.-]*\s*[:=]\s*)"
r"(?![\s\"']*(\$\{|<REDACTED>|$))(.+)$"
)
URL_CRED_RE = re.compile(r"(\w+://[^:/\s]+:)[^@/\s]+@")
# Tipos de recurso que se exportan (además de los values de Helm).
KINDS = [
"deployments.apps", "statefulsets.apps", "daemonsets.apps", "cronjobs.batch",
"services", "ingresses.networking.k8s.io", "persistentvolumeclaims", "configmaps",
"storageclasses.storage.k8s.io",
"ipaddresspools.metallb.io", "l2advertisements.metallb.io",
"clusterpolicies.kyverno.io",
"kafkas.kafka.strimzi.io", "kafkanodepools.kafka.strimzi.io",
"kafkaconnects.kafka.strimzi.io", "kafkaconnectors.kafka.strimzi.io",
"elasticsearches.elasticsearch.k8s.elastic.co", "kibanas.kibana.k8s.elastic.co",
"agents.agent.k8s.elastic.co",
"applications.argoproj.io",
"schedules.velero.io", "backupstoragelocations.velero.io",
"recurringjobs.longhorn.io",
]
SKIP_CONFIGMAPS = {"kube-root-ca.crt", "istio-ca-root-cert", "extension-apiserver-authentication",
"kube-apiserver-legacy-service-account-token-tracking", "cluster-dns"}
SKIP_NAMESPACES = {"kube-node-lease", "kube-public"}
# Componentes instalados con el manifiesto oficial (kubectl apply -f <url>): su despliegue
# se documenta en docs/ con la versión; aquí solo se exporta la configuración propia.
UPSTREAM_NAMESPACES = {"argocd", "cert-manager", "metallb-system", "elastic-system", "longhorn-system"}
OWN_CONFIG_KINDS = {"Application", "IPAddressPool", "L2Advertisement", "RecurringJob"}
# Objetos que crea un operador sin ownerReferences.
SKIP_OBJECTS = {("kube-system", "Service", "prometheus-kube-prometheus-kubelet")}
DROP_META = {"uid", "resourceVersion", "generation", "creationTimestamp",
"managedFields", "selfLink", "finalizers", "ownerReferences"}
DROP_ANNOTATIONS = re.compile(
r"^(kubectl\.kubernetes\.io/last-applied-configuration|deployment\.kubernetes\.io/"
r"|pv\.kubernetes\.io/|volume\.beta\.kubernetes\.io/storage-provisioner"
r"|volume\.kubernetes\.io/storage-provisioner|kubectl\.kubernetes\.io/restartedAt"
r"|metallb\.universe\.tf/ip-allocated-from-pool)"
)
def run(*cmd):
return subprocess.run(cmd, check=True, capture_output=True, text=True).stdout
def redact(obj, key=""):
"""Oculta credenciales de forma recursiva."""
if isinstance(obj, dict):
# Variables de entorno: {name: DB_PASSWORD, value: ...}
if "name" in obj and "value" in obj and isinstance(obj.get("name"), str) \
and SECRET_KEY_RE.search(obj["name"].replace("_", "")) and obj["value"]:
obj = dict(obj, value=REDACTED)
return {k: redact(v, k) for k, v in obj.items()}
if isinstance(obj, list):
return [redact(v, key) for v in obj]
if isinstance(obj, str):
if key and SECRET_KEY_RE.search(key) and not SAFE_KEY_RE.search(key) and obj:
return REDACTED
obj = INLINE_RE.sub(lambda m: m.group(1) + REDACTED, obj)
return URL_CRED_RE.sub(r"\1" + REDACTED + "@", obj)
if key and SECRET_KEY_RE.search(key) and not SAFE_KEY_RE.search(key) \
and obj not in (None, True, False):
return REDACTED
return obj
def managed_elsewhere(meta, argocd_apps):
labels = meta.get("labels") or {}
ann = meta.get("annotations") or {}
if meta.get("ownerReferences"):
return True # lo crea un operador/controlador
if labels.get("app.kubernetes.io/managed-by") == "Helm" or "meta.helm.sh/release-name" in ann:
return True # está en cluster/helm/
if any(k.startswith("objectset.rio.cattle.io") for k in list(ann) + list(labels)):
return True # addon de k3s
if "argocd.argoproj.io/tracking-id" in ann or labels.get("app.kubernetes.io/instance") in argocd_apps:
return True # lo gestiona ArgoCD (repo GitOps)
return False
def clean(obj):
meta = obj["metadata"]
for k in DROP_META:
meta.pop(k, None)
ann = {k: v for k, v in (meta.get("annotations") or {}).items() if not DROP_ANNOTATIONS.match(k)}
if ann:
meta["annotations"] = ann
else:
meta.pop("annotations", None)
obj.pop("status", None)
spec = obj.get("spec") or {}
if obj["kind"] == "Service":
for k in ("clusterIP", "clusterIPs", "ipFamilies", "ipFamilyPolicy", "internalTrafficPolicy"):
spec.pop(k, None)
if obj["kind"] == "PersistentVolumeClaim":
spec.pop("volumeName", None)
spec.pop("volumeMode", None)
tmpl = spec.get("template", {})
tmpl.get("metadata", {}).pop("creationTimestamp", None)
if not tmpl.get("metadata"):
tmpl.pop("metadata", None)
return redact(obj)
def _str_presenter(dumper, value):
# Textos de varias líneas (ficheros de configuración) como bloque literal legible.
style = "|" if "\n" in value else None
if style:
value = "\n".join(line.rstrip() for line in value.splitlines()) + "\n"
return dumper.represent_scalar("tag:yaml.org,2002:str", value, style=style)
yaml.SafeDumper.add_representer(str, _str_presenter)
def dump(path, data, header=None):
path.parent.mkdir(parents=True, exist_ok=True)
text = yaml.safe_dump(data, sort_keys=False, allow_unicode=True, width=120)
path.write_text((header or "") + text)
def export_helm():
releases = json.loads(run("helm", "list", "-A", "-o", "json"))
index = []
for r in sorted(releases, key=lambda r: (r["namespace"], r["name"])):
chart, _, version = r["chart"].rpartition("-")
index.append({"namespace": r["namespace"], "release": r["name"], "chart": chart,
"version": version, "appVersion": r.get("app_version", "")})
values = json.loads(run("helm", "get", "values", "-n", r["namespace"], r["name"], "-o", "json")) or {}
header = (f"# Helm release {r['namespace']}/{r['name']}\n"
f"# Chart: {r['chart']} (app {r.get('app_version', '')})\n"
f"# Values aplicados por el usuario. Credenciales sustituidas por {REDACTED}.\n")
dump(OUT / "helm" / r["namespace"] / f"{r['name']}.yaml", redact(values), header)
dump(OUT / "helm" / "releases.yaml", index,
"# Releases de Helm instaladas en el clúster (generado por scripts/export_cluster.py)\n")
return len(index)
def export_manifests():
apps = json.loads(run("kubectl", "get", "applications.argoproj.io", "-A", "-o", "json"))
argocd_apps = {a["metadata"]["name"] for a in apps["items"]}
count = 0
for kind in KINDS:
try:
items = json.loads(run("kubectl", "get", kind, "-A", "-o", "json"))["items"]
except subprocess.CalledProcessError:
continue # CRD no instalada
for obj in items:
meta = obj["metadata"]
ns = meta.get("namespace", "_cluster")
if ns in SKIP_NAMESPACES or managed_elsewhere(meta, argocd_apps):
continue
if ns in UPSTREAM_NAMESPACES and obj["kind"] not in OWN_CONFIG_KINDS:
continue
if (ns, obj["kind"], meta["name"]) in SKIP_OBJECTS:
continue
if obj["kind"] == "ConfigMap" and meta["name"] in SKIP_CONFIGMAPS:
continue
if obj["kind"] == "Service" and meta["name"] == "kubernetes":
continue
dump(OUT / "manifests" / ns / f"{obj['kind'].lower()}-{meta['name']}.yaml", clean(obj))
count += 1
return count
def export_secret_inventory():
secrets = json.loads(run("kubectl", "get", "secrets", "-A", "-o", "json"))["items"]
inv = {}
for s in secrets:
if s.get("type") in ("helm.sh/release.v1", "kubernetes.io/service-account-token"):
continue
ns = s["metadata"]["namespace"]
inv.setdefault(ns, []).append({"name": s["metadata"]["name"], "type": s.get("type"),
"keys": sorted((s.get("data") or {}).keys())})
dump(OUT / "secrets-inventory.yaml", dict(sorted(inv.items())),
"# Secrets existentes en el clúster: solo nombres y claves, NUNCA valores.\n"
"# Los valores viven únicamente en el clúster (y en el gestor de contraseñas).\n")
LEAK_RE = re.compile(
r"-----BEGIN [A-Z ]*PRIVATE KEY-----|client-key-data|\bghp_[A-Za-z0-9]{20,}|\bxox[bp]-|"
r"\bAKIA[0-9A-Z]{16}\b|\bK10[0-9a-f]{40,}"
)
def leak_scan():
found = [str(p.relative_to(ROOT)) for p in OUT.rglob("*.yaml") if LEAK_RE.search(p.read_text())]
if found:
sys.exit("ERROR: posible credencial en: " + ", ".join(found))
def main():
# Borra solo lo generado; cluster/README.md se conserva.
for generated in ("helm", "manifests"):
shutil.rmtree(OUT / generated, ignore_errors=True)
(OUT / "secrets-inventory.yaml").unlink(missing_ok=True)
n_helm = export_helm()
n_man = export_manifests()
export_secret_inventory()
leak_scan()
print(f"OK: {n_helm} releases de Helm y {n_man} manifiestos exportados en {OUT.relative_to(ROOT)}/")
if __name__ == "__main__":
main()