seguridad.py
#!/usr/bin/env python3
"""Semáforo de seguridad de todo el clúster. Solo lectura: no cambia nada.
Uso: ./scripts/seguridad.py [--resumen] [--informes DIR] [--sin-internet]
Revisa en menos de un minuto:
- Clúster: nodos, versión de k3s y su soporte, certificados, RBAC.
- Cargas: contenedores privilegiados o con acceso al host, imágenes sin
versión fija y contraseñas escritas en variables de entorno.
- Exposición: rutas públicas (Ingress) y servicios NodePort/LoadBalancer.
- Herramientas: Falco (24 h), Kyverno, Trivy Operator y alertas de Prometheus.
- Nodos (por SSH, sin sudo): actualizaciones, reinicio pendiente, SSH,
Tailscale SSH, cortafuegos y puertos abiertos.
- Tailscale: equipos que ofrecen Tailscale SSH y equipos olvidados.
- Copias: Velero, copias del ERP y Longhorn.
Cada punto sale en verde, amarillo o rojo. El informe completo se guarda en
Markdown en DIR (por defecto ./informes, fuera de git). Nunca muestra valores de
Secrets ni de variables: solo nombres.
Código de salida: 0 todo verde, 1 hay amarillos, 2 hay rojos.
Requisitos: kubectl con acceso al clúster, ssh con clave a los nodos (usuario
$SSH_USER, por defecto admin), openssl y, opcional, tailscale.
"""
import argparse
import concurrent.futures
import datetime as dt
import json
import os
import re
import shutil
import subprocess
import sys
import urllib.request
SSH_USER = os.environ.get("SSH_USER", "admin")
VIP = "192.0.2.250"
AHORA = dt.datetime.now(dt.timezone.utc)
# Cargas que necesitan acceso al host por cómo funcionan (almacenamiento, red,
# monitorización, detección). Lo que tenga ese acceso fuera de aquí se avisa.
NS_CON_HOST = {"kube-system", "longhorn-system", "falco", "metallb-system", "logging"}
CARGAS_CON_HOST = {"monitoring/prometheus-prometheus-node-exporter"}
# Falsos positivos conocidos de k3s (el scheduler, el controller-manager y el
# proxy van dentro del binario y Prometheus no los encuentra). Ver pendientes R2.
ALERTAS_IGNORADAS = {"Watchdog", "InfoInhibitor", "KubeSchedulerDown",
"KubeControllerManagerDown", "KubeProxyDown"}
# Avisos de Falco que ya sabemos que son normales: (regla, proceso padre).
# update-notifier es el aviso de actualizaciones del escritorio de Ubuntu.
FALCO_CONOCIDOS = {("Read sensitive file untrusted", "update-notifier")}
# Puertos que es normal ver abiertos fuera de localhost en un nodo k3s.
PUERTOS_ESPERADOS = {22, 111, 2379, 2380, 6443, 7472, 7946, 9100, 10250}
NODEPORTS = range(30000, 32768)
# Fin de soporte de k3s/Kubernetes por si no hay Internet (endoflife.date).
EOL_RESERVA = {"1.34": "2026-10-27"}
UIS_ADMIN = re.compile(r"argocd|grafana|kibana|portainer|jenkins|elasticsearch|"
r"falcosidekick-ui|policy-reporter-ui|longhorn-frontend|"
r"prometheus|alertmanager|minio|pihole", re.I)
VAR_SECRETA = re.compile(r"PASS|SECRET|TOKEN|API_?KEY|CREDENTIAL|PRIVATE", re.I)
VAR_REFERENCIA = re.compile(r"_(FILE|PATH|NAME|DIR)$", re.I)
COLOR = sys.stdout.isatty()
ICONO = {"ok": "🟢", "aviso": "🟡", "mal": "🔴"}
class Semaforo:
def __init__(self):
self.secciones = [] # [(título, [(nivel, texto, [detalles])])]
def seccion(self, titulo):
self.secciones.append((titulo, []))
def _poner(self, nivel, texto, detalles=()):
self.secciones[-1][1].append((nivel, texto, list(detalles)))
def ok(self, texto, detalles=()):
self._poner("ok", texto, detalles)
def aviso(self, texto, detalles=()):
self._poner("aviso", texto, detalles)
def mal(self, texto, detalles=()):
self._poner("mal", texto, detalles)
def cuenta(self):
c = {"ok": 0, "aviso": 0, "mal": 0}
for _, puntos in self.secciones:
for nivel, _, _ in puntos:
c[nivel] += 1
return c
def terminal(self, resumen=False):
c = self.cuenta()
for titulo, puntos in self.secciones:
visibles = [p for p in puntos if not resumen or p[0] != "ok"]
if not visibles:
continue
print(f"\n\033[1m{titulo}\033[0m" if COLOR else f"\n{titulo}")
for nivel, texto, detalles in visibles:
print(f" {ICONO[nivel]} {texto}")
if not resumen:
for d in detalles[:15]:
print(f" · {d}")
if len(detalles) > 15:
print(f" · … y {len(detalles) - 15} más (ver informe)")
print(f"\nResultado: 🟢 {c['ok']} · 🟡 {c['aviso']} · 🔴 {c['mal']}")
def markdown(self):
c = self.cuenta()
fecha = dt.datetime.now().strftime("%Y-%m-%d %H:%M")
out = [f"# Semáforo de seguridad — {fecha}", "",
f"**🟢 {c['ok']} · 🟡 {c['aviso']} · 🔴 {c['mal']}**", "",
"Generado por `scripts/seguridad.py` (solo lectura).", ""]
for titulo, puntos in self.secciones:
out += [f"## {titulo}", ""]
for nivel, texto, detalles in puntos:
out.append(f"- {ICONO[nivel]} {texto}")
out += [f" - {d}" for d in detalles]
out.append("")
return "\n".join(out)
def run(cmd, timeout=60, entrada=None):
"""Ejecuta un comando y devuelve (código, salida). Nunca lanza excepción."""
try:
r = subprocess.run(cmd, input=entrada, capture_output=True, text=True, timeout=timeout)
return r.returncode, r.stdout
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
return 1, str(e)
def kjson(*args, timeout=60):
cod, out = run(["kubectl", *args, "-o", "json"], timeout=timeout)
if cod != 0:
return None
try:
return json.loads(out)
except json.JSONDecodeError:
return None
def fecha_k8s(s):
return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
def dias_hasta(fecha):
return (fecha.date() - AHORA.date()).days
def hace(fecha):
h = (AHORA - fecha).total_seconds() / 3600
return f"hace {h:.0f} h" if h < 48 else f"hace {h / 24:.0f} días"
# ───────────────────────────── Clúster ─────────────────────────────
def eol_k8s(sin_internet):
if not sin_internet:
try:
with urllib.request.urlopen("https://endoflife.date/api/kubernetes.json", timeout=6) as r:
return {c["cycle"]: c["eol"] for c in json.load(r) if isinstance(c.get("eol"), str)}
except Exception:
pass
return dict(EOL_RESERVA)
def certificado(host, puerto):
"""Fecha de caducidad del certificado que sirve host:puerto (o None)."""
try:
s = subprocess.run(["openssl", "s_client", "-connect", f"{host}:{puerto}",
"-servername", host], input="", capture_output=True,
text=True, timeout=8)
x = subprocess.run(["openssl", "x509", "-noout", "-enddate"], input=s.stdout,
capture_output=True, text=True, timeout=5)
m = re.search(r"notAfter=(.+)", x.stdout)
if not m:
return None
return dt.datetime.strptime(m.group(1).strip(), "%b %d %H:%M:%S %Y %Z").replace(tzinfo=dt.timezone.utc)
except Exception:
return None
def revisar_cluster(s, nodos, args):
s.seccion("Clúster")
caidos = [n["metadata"]["name"] for n in nodos
if not any(c["type"] == "Ready" and c["status"] == "True" for c in n["status"]["conditions"])]
if caidos:
s.mal(f"{len(caidos)} de {len(nodos)} nodos no están Ready", caidos)
else:
s.ok(f"{len(nodos)} nodos Ready")
versiones = {n["metadata"]["name"]: n["status"]["nodeInfo"]["kubeletVersion"] for n in nodos}
distintas = sorted(set(versiones.values()))
eol = eol_k8s(args.sin_internet)
for v in distintas:
rama = ".".join(v.lstrip("v").split(".")[:2])
quienes = [n for n, x in versiones.items() if x == v]
fin = eol.get(rama)
if not fin:
s.aviso(f"k3s {v}: no sé cuándo pierde el soporte la rama {rama}", quienes)
continue
d = dias_hasta(dt.datetime.fromisoformat(fin).replace(tzinfo=dt.timezone.utc))
texto = f"k3s {v} en {len(quienes)} nodo(s): la rama {rama} "
if d < 0:
s.mal(texto + f"está SIN SOPORTE desde el {fin}", quienes)
elif d < 45:
cuando = "hoy" if d == 0 else "mañana" if d == 1 else f"en {d} días"
s.aviso(texto + f"pierde el soporte el {fin} ({cuando})", quienes)
else:
s.ok(texto + f"tiene soporte hasta el {fin}", quienes)
if len(distintas) > 1:
s.aviso(f"Versiones de k3s mezcladas: {', '.join(distintas)}")
certs = []
for n in nodos:
ip = next(a["address"] for a in n["status"]["addresses"] if a["type"] == "InternalIP")
certs.append((f"kubelet {n['metadata']['name']}", ip, 10250))
certs.insert(0, ("API (VIP)", VIP, 6443))
with concurrent.futures.ThreadPoolExecutor(8) as ex:
fechas = list(ex.map(lambda c: certificado(c[1], c[2]), certs))
malos, avisos, sin_dato = [], [], []
minimo = None
for (nombre, _, _), f in zip(certs, fechas):
if f is None:
sin_dato.append(nombre)
continue
d = dias_hasta(f)
minimo = d if minimo is None else min(minimo, d)
if d < 7:
malos.append(f"{nombre}: caduca en {d} días")
elif d < 30:
avisos.append(f"{nombre}: caduca en {d} días")
if malos:
s.mal("Certificados a punto de caducar", malos)
if avisos:
s.aviso("Certificados que caducan este mes (k3s los renueva al reiniciar si faltan < 120 días)", avisos)
if not malos and not avisos and minimo is not None:
s.ok(f"Certificados de la API y de los kubelets: el primero caduca en {minimo} días")
if sin_dato:
s.aviso("No he podido leer el certificado de", sin_dato)
cm = kjson("get", "certificates.cert-manager.io", "-A")
if cm:
no_listos = [f"{c['metadata']['namespace']}/{c['metadata']['name']}" for c in cm["items"]
if not any(x["type"] == "Ready" and x["status"] == "True"
for x in c.get("status", {}).get("conditions", []))]
if no_listos:
s.aviso("Certificados de cert-manager que no están listos", no_listos)
else:
s.ok(f"cert-manager: {len(cm['items'])} certificado(s) listos")
crb = kjson("get", "clusterrolebindings")
if crb:
admins = []
for b in crb["items"]:
if b["roleRef"]["name"] != "cluster-admin" or b["metadata"]["name"].startswith("system:"):
continue
for sub in b.get("subjects") or []:
if sub["name"].startswith("system:"):
continue
ns = f"{sub['namespace']}/" if sub.get("namespace") else ""
admins.append(f"{sub['kind']} {ns}{sub['name']} (binding {b['metadata']['name']})")
if admins:
s.aviso(f"{len(admins)} cuenta(s) con cluster-admin: revisa que todas lo necesiten", admins)
else:
s.ok("Nadie fuera del sistema tiene cluster-admin")
# ───────────────────────────── Cargas ─────────────────────────────
def plantillas():
"""(ns/nombre, tipo, spec del pod, réplicas) de cada carga, también las paradas."""
out = []
for tipo in ("deployments", "statefulsets", "daemonsets", "cronjobs"):
d = kjson("get", tipo, "-A") or {"items": []}
for w in d["items"]:
spec = w["spec"]
if tipo == "cronjobs":
spec = spec["jobTemplate"]["spec"]
nombre = f"{w['metadata']['namespace']}/{w['metadata']['name']}"
out.append((nombre, tipo[:-1], spec["template"]["spec"], w["spec"].get("replicas")))
pods = kjson("get", "pods", "-A") or {"items": []}
for p in pods["items"]:
if not p["metadata"].get("ownerReferences"):
out.append((f"{p['metadata']['namespace']}/{p['metadata']['name']}", "pod", p["spec"], None))
return out
def revisar_cargas(s):
s.seccion("Cargas (todos los namespaces)")
cargas = plantillas()
# Solo los nombres de los Secrets, nunca su contenido: una variable cuyo
# valor es el nombre de un Secret es una referencia, no una contraseña.
_, out = run(["kubectl", "get", "secrets", "-A", "--no-headers",
"-o", "custom-columns=N:.metadata.namespace,M:.metadata.name"])
nombres_secret = {tuple(l.split()) for l in out.splitlines() if len(l.split()) == 2}
acceso_host, latest, secretas = [], [], []
for nombre, tipo, spec, replicas in cargas:
ns = nombre.split("/")[0]
parado = " (parado)" if replicas == 0 else ""
contenedores = spec.get("containers", []) + spec.get("initContainers", [])
marcas = []
if any((c.get("securityContext") or {}).get("privileged") for c in contenedores):
marcas.append("privilegiado")
if spec.get("hostNetwork"):
marcas.append("red del host")
if spec.get("hostPID"):
marcas.append("procesos del host")
rutas = [v["hostPath"]["path"] for v in spec.get("volumes", []) if "hostPath" in v]
if rutas:
marcas.append("hostPath " + ",".join(sorted(set(rutas)))[:80])
if marcas and ns not in NS_CON_HOST and nombre not in CARGAS_CON_HOST:
acceso_host.append(f"{nombre}{parado}: {'; '.join(marcas)}")
for c in contenedores:
img = c["image"].split("@")[0]
etiqueta = img.rsplit(":", 1)[1] if ":" in img.rsplit("/", 1)[-1] else ""
if "@sha256:" not in c["image"] and etiqueta in ("", "latest"):
latest.append(f"{nombre}{parado}: {c['image']}")
for e in c.get("env", []):
v = e.get("value")
if (VAR_SECRETA.search(e["name"]) and not VAR_REFERENCIA.search(e["name"])
and v and v.lower() not in ("true", "false", "yes", "no", "0", "1")
and not v.startswith("/") and (ns, v) not in nombres_secret):
secretas.append(f"{nombre}{parado}: {e['name']}")
if acceso_host:
s.aviso(f"{len(acceso_host)} carga(s) con acceso al host fuera de lo esperado", sorted(acceso_host))
else:
s.ok("Solo tienen acceso al host las cargas de sistema (Longhorn, Falco, MetalLB, Elastic Agent, node-exporter)")
if latest:
s.aviso(f"{len(latest)} contenedor(es) con imagen sin versión fija (latest)", sorted(set(latest)))
else:
s.ok("Todas las imágenes tienen versión fija")
if secretas:
s.mal(f"{len(secretas)} contraseña(s) escritas en variables de entorno (deberían ir en un Secret)",
sorted(set(secretas)))
else:
s.ok("Ninguna contraseña escrita en variables de entorno")
s.ok(f"Revisadas {len(cargas)} cargas (Deployments, StatefulSets, DaemonSets, CronJobs y pods sueltos)")
# ───────────────────────────── Exposición ─────────────────────────────
def revisar_exposicion(s):
s.seccion("Exposición")
eps = kjson("get", "endpoints", "-A") or {"items": []}
con_backend = {f"{e['metadata']['namespace']}/{e['metadata']['name']}"
for e in eps["items"] if any(sub.get("addresses") for sub in e.get("subsets") or [])}
ing = kjson("get", "ingress", "-A") or {"items": []}
activas, huerfanas = [], []
for i in ing["items"]:
ns = i["metadata"]["namespace"]
for r in i["spec"].get("rules", []):
svcs = {p["backend"]["service"]["name"] for p in r.get("http", {}).get("paths", [])
if "service" in p.get("backend", {})}
vivo = any(f"{ns}/{x}" in con_backend for x in svcs)
(activas if vivo else huerfanas).append(f"{r.get('host', '*')} → {ns}/{','.join(sorted(svcs))}")
if activas:
s.aviso(f"{len(activas)} ruta(s) públicas con servicio en marcha: revisa que deban estar abiertas", activas)
else:
s.ok("Ninguna ruta pública del clúster tiene servicio en marcha")
if huerfanas:
s.aviso(f"{len(huerfanas)} ruta(s) públicas apuntan a servicios parados: bórralas también en Cloudflare",
huerfanas)
svc = kjson("get", "svc", "-A") or {"items": []}
abiertos, admin = [], []
for x in svc["items"]:
t = x["spec"]["type"]
if t not in ("NodePort", "LoadBalancer"):
continue
nombre = f"{x['metadata']['namespace']}/{x['metadata']['name']}"
ips = ",".join(i.get("ip", "") for i in x.get("status", {}).get("loadBalancer", {}).get("ingress", []))
puertos = ",".join(str(p.get("nodePort") or p["port"]) for p in x["spec"].get("ports", []))
linea = f"{nombre} ({t} {ips + ' ' if ips else ''}{puertos})"
if nombre in con_backend:
abiertos.append(linea)
if UIS_ADMIN.search(nombre):
admin.append(linea)
s.ok(f"{len(abiertos)} servicio(s) NodePort/LoadBalancer en marcha, solo LAN y Tailscale", sorted(abiertos))
if admin:
s.aviso(f"{len(admin)} panel(es) de administración sin TLS ni login único en la LAN/Tailscale",
sorted(admin))
# ───────────────────────────── Herramientas ─────────────────────────────
def revisar_herramientas(s, nodos):
s.seccion("Herramientas de seguridad")
ds = kjson("-n", "falco", "get", "ds", "falco")
if not ds:
s.mal("Falco no está instalado")
else:
st = ds["status"]
cubiertos = {p["spec"].get("nodeName") for p in (kjson("-n", "falco", "get", "pods",
"-l", "app.kubernetes.io/name=falco") or {"items": []})["items"]}
faltan = sorted({n["metadata"]["name"] for n in nodos} - cubiertos)
if faltan:
s.aviso(f"Falco vigila {st.get('numberReady', 0)} de {len(nodos)} nodos", [f"sin Falco: {n}" for n in faltan])
else:
s.ok(f"Falco vigila los {len(nodos)} nodos")
revisar_falco(s)
pol = kjson("get", "clusterpolicies.kyverno.io")
if pol is None:
s.aviso("Kyverno no responde")
else:
modos = []
for p in pol["items"]:
sp = p["spec"]
modo = sp.get("validationFailureAction") or next(
(r.get("validate", {}).get("failureAction") for r in sp.get("rules", [])), None) or "Audit"
modos.append(f"{p['metadata']['name']} ({modo})")
if len(pol["items"]) < 3:
s.aviso(f"Kyverno solo tiene {len(pol['items'])} política(s)", modos)
else:
s.ok(f"Kyverno: {len(pol['items'])} políticas", modos)
rep = kjson("get", "policyreports.wgpolicyk8s.io", "-A") or {"items": []}
fallos = sorted({f"{r['policy']}: {res['namespace'] + '/' if res.get('namespace') else ''}{res['name']}"
for x in rep["items"] for r in x.get("results", []) if r.get("result") == "fail"
for res in r.get("resources", [{}]) if res})
if fallos:
s.aviso(f"{len(fallos)} incumplimiento(s) de las políticas de Kyverno", fallos)
else:
s.ok("Sin incumplimientos de Kyverno")
trivy = kjson("-n", "trivy-system", "get", "deploy", "trivy-operator")
vr = kjson("get", "vulnerabilityreports.aquasecurity.github.io", "-A") or {"items": []}
crit = sum(x["report"]["summary"].get("criticalCount", 0) for x in vr["items"])
altas = sum(x["report"]["summary"].get("highCount", 0) for x in vr["items"])
ultimo = max((fecha_k8s(x["metadata"]["creationTimestamp"]) for x in vr["items"]), default=None)
datos = f"{len(vr['items'])} informes, {crit} críticas y {altas} altas" + (f", el último {hace(ultimo)}" if ultimo else "")
if not trivy or trivy["spec"].get("replicas", 1) == 0:
s.aviso(f"Trivy Operator parado: no hay escaneo continuo de imágenes ({datos})",
["pendiente n.º 14: reactivarlo con límites y fuera de los masters"])
elif crit:
s.mal(f"Trivy: {datos}")
else:
s.ok(f"Trivy: {datos}")
cod, out = run(["kubectl", "get", "--raw", "/api/v1/namespaces/monitoring/services/"
"prometheus-kube-prometheus-prometheus:9090/proxy/api/v1/alerts"])
if cod != 0:
s.aviso("Prometheus no responde: no sé qué alertas hay activas")
else:
alertas = [a for a in json.loads(out)["data"]["alerts"]
if a["state"] == "firing" and a["labels"].get("alertname") not in ALERTAS_IGNORADAS]
def linea(a):
l = a["labels"]
obj = l.get("job_name") or l.get("pod") or ""
donde = l.get("namespace", "") + (f"/{obj}" if obj else "")
return f"{l['alertname']} {donde}".strip()
crit = sorted({linea(a) for a in alertas if a["labels"].get("severity") == "critical"})
warn = sorted({linea(a) for a in alertas if a["labels"].get("severity") == "warning"})
if crit:
s.mal(f"{len(crit)} alerta(s) críticas activas en Prometheus", crit)
if warn:
s.aviso(f"{len(warn)} alerta(s) de aviso activas en Prometheus", warn)
if not crit and not warn:
s.ok("Prometheus sin alertas críticas ni de aviso (sin contar los falsos positivos de k3s)")
s.aviso("Alertmanager no avisa a nadie todavía (R2): estas alertas solo se ven aquí")
def revisar_falco(s):
pods = (kjson("-n", "falco", "get", "pods", "-l", "app.kubernetes.io/name=falco") or {"items": []})["items"]
por_prioridad = {}
reglas = {}
conocidos = 0
def leer(p):
return run(["kubectl", "-n", "falco", "logs", p["metadata"]["name"], "-c", "falco",
"--since=24h"], timeout=90)[1]
with concurrent.futures.ThreadPoolExecutor(6) as ex:
for out in ex.map(leer, pods):
for l in out.splitlines():
if not l.startswith("{"):
continue
try:
e = json.loads(l)
except json.JSONDecodeError:
continue
pr = e.get("priority", "?")
padre = (e.get("output_fields") or {}).get("proc.pname")
if (e.get("rule"), padre) in FALCO_CONOCIDOS:
conocidos += 1
continue
por_prioridad[pr] = por_prioridad.get(pr, 0) + 1
if pr in ("Emergency", "Alert", "Critical", "Error", "Warning"):
k = (pr, e.get("rule", "?"), e.get("hostname", "?"))
reglas[k] = reglas.get(k, 0) + 1
graves = sum(por_prioridad.get(p, 0) for p in ("Emergency", "Alert", "Critical"))
medios = sum(por_prioridad.get(p, 0) for p in ("Error", "Warning"))
resumen = ", ".join(f"{k}: {v}" for k, v in sorted(por_prioridad.items(), key=lambda x: -x[1])) or "ninguno"
if conocidos:
resumen += f"; {conocidos} conocidos y normales"
detalle = [f"{pr} · {regla} · {host} ×{n}" for (pr, regla, host), n in sorted(reglas.items(), key=lambda x: -x[1])]
if graves:
s.mal(f"Falco: {graves} evento(s) críticos en 24 h ({resumen})", detalle)
elif medios:
s.aviso(f"Falco: {medios} evento(s) Error/Warning en 24 h ({resumen})", detalle)
else:
s.ok(f"Falco: nada grave en 24 h ({resumen})")
# ───────────────────────────── Nodos ─────────────────────────────
# Se ejecuta en cada nodo como usuario normal. Solo lee.
SCRIPT_NODO = r"""
. /etc/os-release; echo "os=$PRETTY_NAME"
echo "kernel=$(uname -r)"
echo "upd=$(apt-get -s -o Debug::NoLocking=1 upgrade 2>/dev/null | grep -c '^Inst')"
echo "sec=$(apt-get -s -o Debug::NoLocking=1 upgrade 2>/dev/null | grep '^Inst' | sed 's/^[^(]*(//' | grep -ci -- -security)"
[ -e /var/run/reboot-required ] && echo reboot=1 || echo reboot=0
grep -qs 'Unattended-Upgrade "1"' /etc/apt/apt.conf.d/20auto-upgrades && echo auto=1 || echo auto=0
for f in /etc/ssh/sshd_config.d/*.conf /etc/ssh/sshd_config; do
[ -r "$f" ] && awk 'tolower($1)=="match"{exit} {print}' "$f"
done | grep -iE '^\s*(PermitRootLogin|PasswordAuthentication|KbdInteractiveAuthentication)\s' | sed 's/^\s*/sshd=/'
echo "ufw=$(systemctl is-active ufw 2>/dev/null)"
echo "fail2ban=$(systemctl is-active fail2ban 2>/dev/null)"
if command -v tailscale >/dev/null; then
echo "ts_ver=$(tailscale version 2>/dev/null | head -1)"
echo "ts_ssh=$(tailscale debug prefs 2>/dev/null | grep -o '"RunSSH": [a-z]*' | awk '{print $2}')"
fi
ss -tlnH 2>/dev/null | awk '{print "puerto=" $4}'
"""
def revisar_nodo(nombre, ip, local):
if local:
cod, out = run(["bash", "-s"], entrada=SCRIPT_NODO, timeout=60)
else:
cod, out = run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
f"{SSH_USER}@{ip}", "bash -s"], entrada=SCRIPT_NODO, timeout=60)
if "os=" not in out:
return nombre, None
datos = {"sshd": [], "puerto": []}
for l in out.splitlines():
k, _, v = l.partition("=")
if k in datos:
datos[k].append(v.strip())
else:
datos[k] = v.strip()
return nombre, datos
def opcion_sshd(lineas, clave, defecto):
for l in lineas: # sshd se queda con el primer valor que encuentra
partes = l.split()
if partes and partes[0].lower() == clave.lower() and len(partes) > 1:
return partes[1].lower()
return defecto
def puerto_inesperado(direccion, nfs=False):
host, _, p = direccion.rpartition(":")
if host.startswith(("127.", "[::1]")) or host in ("localhost",) or "%lo" in host:
return False
if "%tailscale0" in host: # puerto de WireGuard de Tailscale
return False
try:
n = int(p)
except ValueError:
return False
if nfs and (n == 2049 or n >= 32768): # servidor NFS: nfsd y puertos RPC dinámicos
return False
return n not in PUERTOS_ESPERADOS and n not in NODEPORTS
def revisar_nodos(s, nodos):
s.seccion("Nodos (por SSH, sin sudo)")
locales = set(run(["hostname"])[1].split())
trabajos = []
for n in nodos:
nombre = n["metadata"]["name"]
ip = next(a["address"] for a in n["status"]["addresses"] if a["type"] == "InternalIP")
trabajos.append((nombre, ip, nombre in locales))
with concurrent.futures.ThreadPoolExecutor(6) as ex:
resultados = dict(ex.map(lambda t: revisar_nodo(*t), trabajos))
sin_acceso = [f"{n} ({ip}): autoriza la clave de este equipo para {SSH_USER} (solo hay SSH con clave, ver docs/seguridad.md)"
for n, ip, _ in trabajos if resultados[n] is None]
if sin_acceso:
s.aviso(f"Sin acceso SSH a {len(sin_acceso)} nodo(s): no se revisa su sistema", sin_acceso)
actualiz, seguridad, reinicio, sin_auto, root, clave, ts_ssh, puertos, cortafuegos, versiones = ([] for _ in range(10))
for nombre, d in resultados.items():
if d is None:
continue
versiones.append(f"{nombre}: {d.get('os')}, kernel {d.get('kernel')}, Tailscale {d.get('ts_ver', '—')}")
upd, sec = int(d.get("upd") or 0), int(d.get("sec") or 0)
if sec:
seguridad.append(f"{nombre}: {sec} de seguridad ({upd} en total)")
elif upd:
actualiz.append(f"{nombre}: {upd}")
if d.get("reboot") == "1":
reinicio.append(nombre)
if d.get("auto") != "1":
sin_auto.append(nombre)
r = opcion_sshd(d["sshd"], "PermitRootLogin", "prohibit-password")
if r == "yes":
root.append(f"{nombre}: PermitRootLogin yes")
if opcion_sshd(d["sshd"], "PasswordAuthentication", "yes") == "yes":
clave.append(nombre)
if d.get("ts_ssh") == "true":
ts_ssh.append(nombre)
if d.get("ufw") != "active":
cortafuegos.append(f"{nombre}: ufw {d.get('ufw') or 'no instalado'}, fail2ban {d.get('fail2ban') or 'no instalado'}")
nfs = any(p.endswith(":2049") for p in d["puerto"])
raros = sorted({p for p in d["puerto"] if puerto_inesperado(p, nfs)})
if raros:
puertos.append(f"{nombre}: {' '.join(raros)}")
revisados = sum(1 for d in resultados.values() if d)
s.ok(f"Sistema revisado en {revisados} de {len(nodos)} nodos", sorted(versiones))
(s.mal if seguridad else s.ok)(
"Actualizaciones de seguridad pendientes" if seguridad else "Sin actualizaciones de seguridad pendientes", seguridad)
if actualiz:
s.aviso("Actualizaciones normales pendientes", actualiz)
if reinicio:
s.aviso("Nodos que piden reinicio (kernel o librerías nuevas)", reinicio)
if sin_auto:
s.aviso("Nodos sin actualizaciones automáticas de seguridad (unattended-upgrades)", sin_auto)
(s.mal if root else s.ok)("SSH permite entrar como root con contraseña" if root
else "SSH no deja entrar como root con contraseña", root)
if clave:
s.aviso("SSH acepta contraseña (mejor solo clave; confirma antes que entras con clave)", sorted(clave))
(s.mal if ts_ssh else s.ok)("Tailscale SSH activado (fallos «Tailworm» sin parche)" if ts_ssh
else "Tailscale SSH desactivado en los nodos revisados", ts_ssh)
if cortafuegos:
s.aviso("Nodos sin cortafuegos activo (normal en k3s si la LAN es de confianza)", sorted(cortafuegos))
if puertos:
s.aviso("Puertos abiertos fuera de los habituales de k3s: revisa qué son", sorted(puertos))
else:
s.ok("Solo están abiertos los puertos habituales de k3s")
def revisar_tailscale(s):
if not shutil.which("tailscale"):
return
cod, out = run(["tailscale", "status", "--json"])
if cod != 0:
return
s.seccion("Tailscale (toda la red)")
st = json.loads(out)
equipos = list(st.get("Peer", {}).values()) + [st["Self"]]
con_ssh = [f"{p['HostName']} ({p['DNSName'].split('.')[0]})" for p in equipos if p.get("sshHostKeys")]
(s.mal if con_ssh else s.ok)(
"Equipos que ofrecen Tailscale SSH" if con_ssh else f"Ningún equipo de la red ofrece Tailscale SSH ({len(equipos)} equipos)",
con_ssh)
olvidados = []
for p in equipos:
if p.get("Online") or not p.get("LastSeen") or p["LastSeen"].startswith("0001"):
continue
visto = fecha_k8s(p["LastSeen"])
if (AHORA - visto).days > 60:
olvidados.append(f"{p['DNSName'].split('.')[0]}: sin conectarse {hace(visto)}")
if olvidados:
s.aviso("Equipos olvidados en la red: bórralos en la consola (Machines)", olvidados)
s.ok(f"Versión de Tailscale en este equipo: {st.get('Version', '?').split('-')[0]}"
" — vigila https://tailscale.com/security-bulletins")
# ───────────────────────────── Copias ─────────────────────────────
def revisar_copias(s):
s.seccion("Copias y almacenamiento")
b = kjson("-n", "velero", "get", "backups.velero.io")
if b and b["items"]:
ult = max(b["items"], key=lambda x: x["metadata"]["creationTimestamp"])
f = fecha_k8s(ult["metadata"]["creationTimestamp"])
fase = ult.get("status", {}).get("phase", "?")
texto = f"Velero: última copia {ult['metadata']['name']} {fase}, {hace(f)}"
if fase != "Completed":
s.mal(texto)
elif (AHORA - f).days >= 2:
s.aviso(texto)
else:
s.ok(texto)
else:
s.mal("Velero: no hay ninguna copia")
for fichero, nombre, max_h in (("ultima-copia", "copia del ERP", 36),
("ultima-secundaria", "copia secundaria del ERP (worker-3)", 36),
("ultima-prueba", "prueba de restauración del ERP", 9 * 24)):
ruta = f"/var/lib/erp-copias/{fichero}"
if not os.path.exists(ruta):
continue
f = dt.datetime.fromisoformat(open(ruta).read().strip()).astimezone(dt.timezone.utc)
h = (AHORA - f).total_seconds() / 3600
(s.ok if h <= max_h else s.mal)(f"Última {nombre}: {hace(f)}")
vols = kjson("-n", "longhorn-system", "get", "volumes.longhorn.io") or {"items": []}
malos = [f"{v['status'].get('kubernetesStatus', {}).get('pvcName') or v['metadata']['name']}: "
f"{v['status'].get('robustness')}" for v in vols["items"]
if v["status"].get("state") == "attached" and v["status"].get("robustness") != "healthy"]
(s.aviso if malos else s.ok)(f"Longhorn: {len(malos)} volumen(es) en uso no healthy" if malos
else "Longhorn: todos los volúmenes en uso están healthy", malos)
def main():
ap = argparse.ArgumentParser(description="Semáforo de seguridad del clúster (solo lectura)")
ap.add_argument("--resumen", action="store_true", help="solo amarillos y rojos, sin detalles")
ap.add_argument("--informes", default="informes", help="carpeta del informe Markdown (por defecto ./informes)")
ap.add_argument("--sin-informe", action="store_true", help="no guarda el informe")
ap.add_argument("--sin-internet", action="store_true", help="no consulta endoflife.date")
args = ap.parse_args()
nodos = kjson("get", "nodes")
if not nodos:
sys.exit("kubectl no responde: no puedo revisar el clúster")
nodos = nodos["items"]
s = Semaforo()
for paso in (lambda: revisar_cluster(s, nodos, args), lambda: revisar_cargas(s),
lambda: revisar_exposicion(s), lambda: revisar_herramientas(s, nodos),
lambda: revisar_nodos(s, nodos), lambda: revisar_tailscale(s),
lambda: revisar_copias(s)):
try:
paso()
except Exception as e: # una parte rota no debe tapar el resto
s.aviso(f"Error al revisar esta parte: {type(e).__name__}: {e}")
s.terminal(resumen=args.resumen)
if not args.sin_informe:
os.makedirs(args.informes, mode=0o700, exist_ok=True)
ruta = os.path.join(args.informes, f"seguridad-{dt.datetime.now():%Y-%m-%d-%H%M}.md")
with open(ruta, "w") as f:
f.write(s.markdown())
os.chmod(ruta, 0o600)
print(f"Informe: {ruta}")
c = s.cuenta()
sys.exit(2 if c["mal"] else 1 if c["aviso"] else 0)
if __name__ == "__main__":
main()