#!/usr/bin/env python3
"""Control General — sonda. Se ejecuta EN cada máquina y escribe un JSON en
stdout con su estado: sistema, discos físicos (SMART), montajes, mdadm, Docker
y red. Solo lectura. Solo librería estándar. Tiene que correr con Python 3.8
(boira) y como root (smartctl, journalctl).

Cómo llega aquí:
  - desarrollo: el colector la manda por stdin (`ssh … 'sudo -n python3 - --smart'`)
  - producción: instalada como /usr/local/bin/cg-sonda y forzada en
    authorized_keys (`command="/usr/local/bin/cg-sonda"`); los argumentos del
    colector llegan por SSH_ORIGINAL_COMMAND.

Argumentos: --smart  -> además de lo básico, consulta SMART (es lo lento).
            --sin-docker / --sin-red  -> para pruebas.
            Fase 3 (por máquina, según su config): --ping ip,ip,…  --wireguard
            --nfs  --cert <sni>  --backup-state <ruta>  --ip-publica [dominio]
            v6: --actualizaciones -> imágenes de los contenedores (tags, digests,
            versión, compose), paquetes apt pendientes y sistema operativo.
            v7: --accion=actualizar_sonda  (por stdin {"sonda", "firma"}) -> se
            sustituye a sí misma por una sonda nueva FIRMADA y de versión mayor.
            v8: sistema.kernel_nuevo -> kernel instalado más nuevo que el que corre
            (en Debian no siempre se crea /var/run/reboot-required).
            v9: lanzada como prueba (cg-sonda.nueva) no hace caso de SSH_ORIGINAL_COMMAND:
            con la clave forzada lo heredaba de la sonda vieja y la v7->v8 nunca se instaló.
Los discos dormidos NO se despiertan (smartctl -n standby): se devuelven con
"dormido": true y el colector conserva el último SMART que les leyó.
"""
import glob
import json
import os
import re
import shlex
import socket
import subprocess
import sys
import threading
import time

SONDA_VERSION = 14
TIMEOUT_CMD = 20


# ----------------------------------------------------------------- utilidades
def run(cmd, timeout=TIMEOUT_CMD, text=True, sin=()):
    """(rc, stdout, stderr). rc = -1 si no existe el binario o se pasa de tiempo.
    sin: variables de entorno que NO se pasan al programa."""
    try:
        # LC_ALL=C: salidas parseables (boira en es_ES rechaza "-i 0.3" y pone comas)
        env = {k: v for k, v in dict(os.environ, LC_ALL="C", LANG="C").items() if k not in sin}
        p = subprocess.run(cmd, capture_output=True, timeout=timeout, env=env)
        out = p.stdout.decode("utf-8", "replace") if text else p.stdout
        return p.returncode, out, p.stderr.decode("utf-8", "replace")
    except (FileNotFoundError, PermissionError) as e:
        return -1, "", str(e)
    except subprocess.TimeoutExpired:
        return -1, "", "timeout"


def which(name):
    for d in ("/usr/sbin", "/sbin", "/usr/bin", "/bin", "/usr/local/sbin", "/usr/local/bin"):
        p = os.path.join(d, name)
        if os.access(p, os.X_OK):
            return p
    return None


def leer(path, default=""):
    try:
        with open(path, "r", errors="replace") as f:
            return f.read()
    except OSError:
        return default


def con_timeout(fn, seconds, default=None):
    """Ejecuta fn en un hilo; si no acaba en `seconds`, devuelve default."""
    box = {}

    def _w():
        try:
            box["v"] = fn()
        except Exception as e:  # noqa: BLE001
            box["e"] = str(e)

    t = threading.Thread(target=_w, daemon=True)
    t.start()
    t.join(seconds)
    if t.is_alive():
        return default, "timeout"
    if "e" in box:
        return default, box["e"]
    return box.get("v"), None


# -------------------------------------------------------------------- sistema
def _cpu_sample():
    for line in leer("/proc/stat").splitlines():
        if line.startswith("cpu "):
            v = [int(x) for x in line.split()[1:]]
            idle = v[3] + (v[4] if len(v) > 4 else 0)
            return sum(v), idle
    return 0, 0


def sistema(cpu_ini):
    d = {"hostname": socket.gethostname()}
    for line in leer("/etc/os-release").splitlines():
        if line.startswith("PRETTY_NAME="):
            d["so"] = line.split("=", 1)[1].strip().strip('"')
    d["kernel"] = os.uname().release
    d["arquitectura"] = os.uname().machine
    try:
        d["uptime_s"] = int(float(leer("/proc/uptime").split()[0]))
    except (ValueError, IndexError):
        d["uptime_s"] = None
    try:
        d["carga"] = list(os.getloadavg())
    except OSError:
        d["carga"] = None
    d["hilos"] = os.cpu_count()
    tot2, idle2 = _cpu_sample()
    tot1, idle1 = cpu_ini
    dt = tot2 - tot1
    d["cpu_pct"] = round(100.0 * (1 - (idle2 - idle1) / dt), 1) if dt > 0 else None

    mem = {}
    for line in leer("/proc/meminfo").splitlines():
        k, _, v = line.partition(":")
        try:
            mem[k.strip()] = int(v.split()[0]) * 1024
        except (ValueError, IndexError):
            pass
    d["mem_total"] = mem.get("MemTotal")
    d["mem_disponible"] = mem.get("MemAvailable")
    if d["mem_total"] and d["mem_disponible"] is not None:
        d["mem_pct"] = round(100.0 * (1 - d["mem_disponible"] / d["mem_total"]), 1)
    d["swap_total"] = mem.get("SwapTotal")
    d["swap_libre"] = mem.get("SwapFree")

    d["temperaturas"] = temperaturas()
    d["unidades_fallidas"] = unidades_fallidas()
    d["kernel_nuevo"] = kernel_nuevo()
    d["reinicio_pendiente"] = os.path.exists("/var/run/reboot-required") or bool(d["kernel_nuevo"])
    d["errores_kernel_1h"] = errores_kernel()
    return d


def _orden_kernel(v):
    """6.12.111+deb13-amd64 -> (6, 12, 111, 13): para ordenar versiones de kernel."""
    return tuple(int(x) for x in re.findall(r"\d+", v))


def kernel_nuevo(boot="/boot", en_marcha=None):
    """(v8) El kernel instalado más nuevo, si no es el que está en marcha; si no, None.
    Lo da /boot/vmlinuz-<versión> (Debian y Ubuntu). En Debian, apt no siempre deja
    /var/run/reboot-required al instalar un kernel: visto en el NUC-red el 2026-10-01."""
    en_marcha = en_marcha or os.uname().release
    try:
        versiones = [n[len("vmlinuz-"):] for n in os.listdir(boot) if n.startswith("vmlinuz-")]
    except OSError:
        return None
    if not versiones:
        return None
    nuevo = max(versiones, key=_orden_kernel)
    return nuevo if _orden_kernel(nuevo) > _orden_kernel(en_marcha) else None


def temperaturas():
    """[{nombre, grados, high, crit}] — sensors -j si existe; si no, thermal_zone."""
    out = []
    if which("sensors"):
        rc, s, _ = run(["sensors", "-j"], timeout=10)
        if rc == 0:
            try:
                data = json.loads(s)
            except ValueError:
                data = {}
            for chip, feats in data.items():
                if not isinstance(feats, dict):
                    continue
                for feat, vals in feats.items():
                    if not isinstance(vals, dict):
                        continue
                    cur = high = crit = None
                    for k, v in vals.items():
                        # v14: solo tempN_*; antes entraban fanN_input (rpm: «fan1 a 158 °C» ❌) e inN_input (voltios)
                        if not k.startswith("temp"):
                            continue
                        if k.endswith("_input"):
                            cur = v
                        elif k.endswith("_max") or k.endswith("_high"):
                            high = v
                        elif k.endswith("_crit"):
                            crit = v
                    if cur is None or cur < -100 or cur > 200:
                        continue  # sensores fantasma (-128 °C en el Fujitsu)
                    # límite 0 o negativo = sin límite (SYSTIN del Fujitsu trae high -1 y avisaba siempre)
                    high = high if (high is not None and high > 0) else None
                    crit = crit if (crit is not None and crit > 0) else None
                    out.append({"chip": chip, "nombre": feat, "grados": round(cur, 1),
                                "high": high, "crit": crit})
    if not out:
        for z in sorted(glob.glob("/sys/class/thermal/thermal_zone*")):
            tipo = leer(os.path.join(z, "type")).strip()
            try:
                t = int(leer(os.path.join(z, "temp")).strip()) / 1000.0
            except ValueError:
                continue
            if t < -100 or t > 200:
                continue
            out.append({"chip": "thermal", "nombre": tipo or os.path.basename(z),
                        "grados": round(t, 1), "high": None, "crit": None})
    return out


def unidades_fallidas():
    rc, s, _ = run(["systemctl", "--failed", "--plain", "--no-legend", "--no-pager"], timeout=10)
    if rc != 0:
        return []
    return [line.split()[0] for line in s.splitlines() if line.strip()]


# El driver e1000e (tarjetas Intel I219 de los NUC y del Fujitsu) escribe esto cuando
# la cola de transmisión se atasca; después la interfaz deja de enviar y la máquina
# sigue viva pero sin red. Visto el 2026-09-19 en el NUC-red (185 veces seguidas).
# El otro patrón es el aviso genérico del kernel para cualquier tarjeta.
_RE_NIC_HANG = re.compile(
    r"(?:(\S+): Detected Hardware Unit Hang"
    r"|NETDEV WATCHDOG: (\S+) \(\S+\): transmit queue \d+ timed out)")


# Líneas de continuación del mensaje del e1000e: unas van con sangría y otras NO
# (`MAC Status`, `PHY Status`, `PCI Status`...), así que la sangría sola no basta.
_RE_NIC_CONT = re.compile(
    r"^\s*(TDH|TDT|next_to_use|next_to_clean|next_to_watch|buffer_info|time_stamp|jiffies|"
    r"MAC Status|PHY (Extended |1000BASE-T )?Status|PCI Status)\b")


def _lineas_nic(texto, ifaces):
    """(nº de avisos del driver, líneas que NO son del cuelgue) de un texto de journalctl.
    Un mensaje del kernel puede ocupar varias líneas (el del e1000e ocupa ~15; unas van
    con sangría y otras no): se descuentan para saber si hubo OTROS errores."""
    n_hang, otros = 0, 0
    for x in texto.splitlines():
        if not x.strip():
            continue
        m = _RE_NIC_HANG.search(x)
        if m:
            n_hang += 1
            ifaces.add(m.group(1) or m.group(2))
        elif x[:1] in (" ", "\t") or _RE_NIC_CONT.match(x):
            continue            # continuación de un mensaje multilínea
        else:
            otros += 1
    return n_hang, otros


def errores_kernel():
    rc, s, _ = run(["journalctl", "-k", "-p", "err", "--since", "-1h", "--no-pager", "-q",
                    "-o", "cat"], timeout=15)
    if rc != 0:
        return None
    lineas = [x for x in s.splitlines() if x.strip()]
    res = {"n": len(lineas), "ultimas": lineas[-3:]}
    ifaces = set()
    n_hang, otros = _lineas_nic(s, ifaces)
    # `journalctl -k` mira SOLO el arranque actual, y la salida habitual de un cuelgue de
    # la tarjeta es reiniciar: sin mirar también el arranque anterior, el aviso desaparecía
    # justo al reiniciar. Solo cuenta lo de la última hora; sin diario persistente falla
    # y se ignora.
    rc2, s2, _ = run(["journalctl", "-k", "-b", "-1", "-p", "err", "--since", "-1h",
                      "--no-pager", "-q", "-o", "cat"], timeout=15)
    n_prev = 0
    if rc2 == 0 and s2.strip():
        n_prev, _otros = _lineas_nic(s2, ifaces)
    if n_hang or n_prev:
        res["nic_hang"] = {"n": n_hang + n_prev, "ifaces": sorted(ifaces),
                           "previo": n_hang == 0 and n_prev > 0}
        res["otros"] = otros
    return res


# --------------------------------------------------------------------- discos
def _lsblk():
    rc, s, _ = run(["lsblk", "-J", "-b", "-o",
                    "NAME,KNAME,TYPE,SIZE,MODEL,SERIAL,TRAN,MOUNTPOINT,FSTYPE,ROTA,PKNAME"])
    if rc != 0:
        # lsblk viejo (boira 20.04 lo tiene bien; por si acaso)
        rc, s, _ = run(["lsblk", "-J", "-b", "-o",
                        "NAME,KNAME,TYPE,SIZE,MODEL,SERIAL,TRAN,MOUNTPOINT,FSTYPE,ROTA"])
    try:
        return json.loads(s).get("blockdevices", [])
    except ValueError:
        return []


def _montajes_de(nodo, acc):
    mp = nodo.get("mountpoint")
    if mp:
        acc.append(mp)
    for h in nodo.get("children", []) or []:
        _montajes_de(h, acc)
    return acc


def _etiqueta(montajes):
    """Nombre de la casa: /media/Boira05 -> Boira05; / -> sistema."""
    for mp in montajes:
        if mp == "/":
            return "sistema"
    for mp in montajes:
        if mp.startswith("/media/") or mp.startswith("/mnt/"):
            return mp.split("/", 2)[2].split("/")[0]
    return montajes[0] if montajes else ""


def _smart(dev):
    """SMART de un disco sin despertarlo. Devuelve dict o {"dormido": True}."""
    if not which("smartctl"):
        return {"error": "sin smartctl"}
    cmd = ["smartctl", "-j", "-n", "standby", "-H", "-A", "-l", "selftest", dev]
    rc, s, err = run(cmd, timeout=40)
    try:
        j = json.loads(s) if s.strip() else {}
    except ValueError:
        j = {}
    # bit 1 del exit status = disco en standby (con -n standby)
    if rc > 0 and (rc & 2):
        return {"dormido": True}
    if not j or (rc > 0 and (rc & 1)):
        # p.ej. USB sin puente SAT reconocido: segundo intento forzando sat
        rc2, s2, _ = run(cmd[:1] + ["-d", "sat"] + cmd[1:], timeout=40)
        try:
            j2 = json.loads(s2) if s2.strip() else {}
        except ValueError:
            j2 = {}
        if j2.get("ata_smart_attributes") or j2.get("smart_status"):
            j = j2
        elif not j:
            msgs = [m.get("string", "") for m in (j.get("smartctl", {}).get("messages") or [])]
            return {"error": "; ".join(msgs) or err.strip() or "smartctl rc=%s" % rc}
    r = {"dormido": False}
    # serie y modelo REALES del disco: en los USB, lsblk enseña los del puente
    # (los cinco USB3.0_DISKxx de boira comparten serie)
    if j.get("serial_number"):
        r["serie"] = str(j["serial_number"]).strip()
    if j.get("model_name"):
        r["modelo"] = str(j["model_name"]).strip()
    st = j.get("smart_status", {})
    if "passed" in st:
        r["salud_ok"] = bool(st["passed"])
    if j.get("temperature", {}).get("current") is not None:
        r["temperatura"] = j["temperature"]["current"]
    if j.get("power_on_time", {}).get("hours") is not None:
        r["horas"] = j["power_on_time"]["hours"]
    attrs = {}
    for a in (j.get("ata_smart_attributes", {}).get("table") or []):
        try:
            attrs[int(a["id"])] = int(a["raw"]["value"])
        except (KeyError, TypeError, ValueError):
            continue
    if attrs:
        r["reasignados"] = attrs.get(5)
        r["pendientes"] = attrs.get(197)
        r["incorregibles"] = attrs.get(198)
        r["errores_crc"] = attrs.get(199)
        r["arranques"] = attrs.get(12)
    nv = j.get("nvme_smart_health_information_log")
    if nv:
        r["nvme"] = {"desgaste_pct": nv.get("percentage_used"),
                     "errores_medio": nv.get("media_errors"),
                     "aviso_critico": nv.get("critical_warning")}
    st_test = (j.get("ata_smart_data", {}).get("self_test", {}).get("status") or {})
    if st_test.get("remaining_percent") is not None:
        r["test_en_curso"] = "%s%% restante" % st_test["remaining_percent"]
    tests = (j.get("ata_smart_self_test_log", {}).get("standard", {}).get("table") or [])
    if tests:
        t = tests[0]
        r["ultimo_test"] = {"tipo": t.get("type", {}).get("string"),
                            "resultado": t.get("status", {}).get("string"),
                            "ok": bool(t.get("status", {}).get("passed", True)),
                            "horas": t.get("lifetime_hours")}
        if r.get("horas") is not None and t.get("lifetime_hours") is not None:
            r["ultimo_test"]["hace_horas"] = r["horas"] - t["lifetime_hours"]
    msgs = [m.get("string", "") for m in (j.get("smartctl", {}).get("messages") or [])
            if m.get("severity") == "error"]
    if msgs and len(r) == 1:
        r["error"] = "; ".join(msgs)
    return r


def _serie_sysfs(kname):
    """Serie del disco cuando lsblk no la da: página VPD 0x80, wwid del kernel y,
    si no, los enlaces de /dev/disk/by-id. Sin tocar el disco: no despierta dormidos."""
    for base in ("/sys/block/%s/device" % kname, "/sys/block/%s" % kname):
        try:
            with open(base + "/vpd_pg80", "rb") as f:
                b = f.read()
            s = b[4:].decode("ascii", "replace").strip(" \x00\t\n")
            if s:
                return s
        except OSError:
            pass
    for path in ("/sys/block/%s/device/wwid" % kname, "/sys/block/%s/wwid" % kname):
        s = " ".join(leer(path).split())
        if s:
            return s
    # enlaces de udev: ata-MODELO_SERIE (SAT a través de puentes USB) y wwn-0x…
    try:
        enlaces = os.listdir("/dev/disk/by-id")
    except OSError:
        enlaces = []
    mios = [e for e in enlaces if "-part" not in e
            and os.path.basename(os.path.realpath("/dev/disk/by-id/" + e)) == kname]
    for pref in ("ata-", "nvme-", "wwn-"):
        for e in sorted(mios):
            if e.startswith(pref):
                return e.rsplit("_", 1)[-1] if pref != "wwn-" else e[len(pref):]
    return ""


def discos(con_smart):
    out = []
    for n in _lsblk():
        if n.get("type") != "disk":
            continue
        kname = n.get("kname") or n.get("name")
        if not kname or kname.startswith(("loop", "zram", "ram", "sr", "fd")):
            continue
        montajes = _montajes_de(n, [])
        d = {"dev": "/dev/" + kname, "nombre": kname, "modelo": (n.get("model") or "").strip(),
             "serie": (n.get("serial") or "").strip(), "bytes": n.get("size"),
             "bus": n.get("tran") or "", "rotacional": bool(int(n.get("rota") or 0)),
             "montajes": montajes, "etiqueta": _etiqueta(montajes)}
        if not d["serie"]:
            d["serie_sys"] = _serie_sysfs(kname)
        if con_smart:
            d["smart"] = _smart(d["dev"])
        out.append(d)
    # id estable e independiente de si la pasada lleva SMART: la serie de lsblk
    # (o, si lsblk no la da, la de sysfs) si es única; si varios discos la
    # comparten (puentes USB), la etiqueta del montaje; y si tampoco (miembros
    # de un md), etiqueta/kname. Si el id cambia, el colector lo empareja.
    series = {}
    for d in out:
        s = d["serie"] or d.get("serie_sys") or ""
        series[s] = series.get(s, 0) + 1
    etiquetas = {}
    for d in out:
        etiquetas[d["etiqueta"]] = etiquetas.get(d["etiqueta"], 0) + 1
    for d in out:
        s = d["serie"] or d.get("serie_sys") or ""
        if s and series[s] == 1:
            d["id"] = s
        elif d["etiqueta"] and etiquetas[d["etiqueta"]] == 1:
            d["id"] = d["etiqueta"]
        else:
            d["id"] = (d["etiqueta"] + "/" if d["etiqueta"] else "") + d["nombre"]
    return out


def mdadm():
    arrays = []
    texto = leer("/proc/mdstat")
    cur = None
    for line in texto.splitlines():
        m = re.match(r"^(md\d+)\s*:\s*(\w+)\s+(\S+)\s+(.*)$", line)
        if m:
            cur = {"nombre": m.group(1), "estado": m.group(2), "nivel": m.group(3),
                   "miembros": re.findall(r"(\w+)\[\d+\]", m.group(4)), "degradado": False}
            arrays.append(cur)
            continue
        if cur and "blocks" in line:
            mm = re.search(r"\[([U_]+)\]", line)
            if mm:
                cur["disposicion"] = mm.group(1)
                cur["degradado"] = "_" in mm.group(1)
            if "resync" in line or "recovery" in line or "check" in line:
                cur["operacion"] = line.strip()
    return arrays


FS_REALES = {"ext2", "ext3", "ext4", "xfs", "btrfs", "vfat", "exfat", "ntfs", "f2fs",
             "nfs", "nfs4", "cifs", "fuse.sshfs", "zfs"}


def montajes():
    out = []
    vistos = set()
    autofs = []
    for line in leer("/proc/mounts").splitlines():
        p = line.split()
        if len(p) < 3:
            continue
        dev, mp, fs = p[0], p[1].replace("\\040", " "), p[2]
        if fs == "autofs":
            autofs.append((dev, mp))
            continue
        if fs not in FS_REALES or mp in vistos:
            continue
        if mp.startswith(("/var/lib/docker", "/snap/", "/boot/efi", "/run/")):
            continue
        vistos.add(mp)
        m = {"punto": mp, "dispositivo": dev, "tipo": "nfs" if fs.startswith("nfs") else
             ("cifs" if fs == "cifs" else "local"), "fs": fs,
             "etiqueta": _etiqueta([mp])}

        def _st(path=mp):
            v = os.statvfs(path)
            return v.f_blocks * v.f_frsize, v.f_bavail * v.f_frsize, v.f_bfree * v.f_frsize

        r, err = con_timeout(_st, 5)
        if r is None:
            m["rancio"] = True
            m["error"] = err
        else:
            total, avail, free = r
            m["rancio"] = False
            m["bytes"] = total
            m["libres"] = avail
            usados = total - free
            m["usados"] = usados
            m["pct"] = round(100.0 * usados / (usados + avail), 1) if (usados + avail) else 0.0
        out.append(m)
    # Automontaje (x-systemd.automount) en espera: solo está el autofs, y el montaje
    # de verdad llega al abrir la carpeta (y se suelta tras el idle-timeout). Cuenta
    # como presente; sin statvfs, que lo montaría.
    fstab = {}
    for line in leer("/etc/fstab").splitlines():
        p = line.split()
        if len(p) >= 3 and not p[0].startswith("#"):
            fstab[p[1].replace("\\040", " ")] = (p[0], p[2])
    for dev, mp in autofs:
        if mp in vistos or mp not in fstab:
            continue
        vistos.add(mp)
        dev, fs = fstab[mp]
        out.append({"punto": mp, "dispositivo": dev, "tipo": "nfs" if fs.startswith("nfs") else
                    ("cifs" if fs == "cifs" else "local"), "fs": fs,
                    "etiqueta": _etiqueta([mp]), "automontaje": True, "rancio": False})
    return out


# --------------------------------------------------------------------- docker
def docker():
    sock = "/var/run/docker.sock"
    if not os.path.exists(sock):
        return {"disponible": False}
    try:
        lista = _docker_get("/containers/json?all=1")
    except Exception as e:  # noqa: BLE001
        return {"disponible": True, "error": str(e)}
    conts = []
    for c in lista:
        status = c.get("Status") or ""
        low = status.lower()
        if "(healthy)" in low:
            health = "healthy"
        elif "(unhealthy)" in low:
            health = "unhealthy"
        elif "health: starting" in low:
            health = "starting"
        else:
            health = ""
        conts.append({"nombre": (c.get("Names") or ["?"])[0].lstrip("/"),
                      "imagen": c.get("Image"), "estado": c.get("State"),
                      "status": status, "health": health, "creado": c.get("Created")})
    conts.sort(key=lambda x: x["nombre"])
    return {"disponible": True, "contenedores": conts}


# ---------------------------------------------------- actualizaciones (v6)
# Solo lectura: qué imagen lleva cada contenedor y qué paquetes del sistema
# esperan. La comparación con el registro la hace el COLECTOR, no la máquina.
# No se lanza `apt update`: se leen las listas que dejó el apt-daily de la máquina
# y se dice su antigüedad.
_ETQ_VERSION = "org.opencontainers.image.version"
_ETQ_COMPOSE = {"compose_dir": "com.docker.compose.project.working_dir",
                "compose_proyecto": "com.docker.compose.project",
                "compose_servicio": "com.docker.compose.service",
                "compose_ficheros": "com.docker.compose.project.config_files"}


def _docker_get(path, timeout=10):
    import http.client
    conn = http.client.HTTPConnection("localhost", timeout=timeout)
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    s.settimeout(timeout)
    s.connect("/var/run/docker.sock")
    conn.sock = s
    conn.request("GET", path)
    r = conn.getresponse()
    data = r.read()
    conn.close()
    if r.status != 200:
        raise RuntimeError("docker api %s -> %s" % (path, r.status))
    return json.loads(data)


def imagenes_docker():
    """Por contenedor: la referencia con la que se creó, el id de imagen, el compose
    del que viene y, por imagen, sus etiquetas y digests de registro."""
    if not os.path.exists("/var/run/docker.sock"):
        return {"disponible": False}
    try:
        lista = _docker_get("/containers/json?all=1")
        motor = _docker_get("/version").get("Version")
    except Exception as e:  # noqa: BLE001
        return {"disponible": True, "error": str(e)}
    conts, imgs = [], {}
    for c in lista:
        etq = c.get("Labels") or {}
        iid = c.get("ImageID") or ""
        e = {"nombre": (c.get("Names") or ["?"])[0].lstrip("/"),
             "ref": c.get("Image"), "imagen_id": iid, "estado": c.get("State")}
        for k, v in _ETQ_COMPOSE.items():
            if etq.get(v):
                e[k] = etq[v]
        conts.append(e)
        if iid and iid not in imgs:
            try:
                i = _docker_get("/images/%s/json" % iid)
            except Exception as ex:  # noqa: BLE001
                imgs[iid] = {"error": str(ex)}
                continue
            ietq = (i.get("Config") or {}).get("Labels") or {}
            imgs[iid] = {"tags": i.get("RepoTags") or [], "digests": i.get("RepoDigests") or [],
                         "version": ietq.get(_ETQ_VERSION), "creada": i.get("Created"),
                         "fuente": ietq.get("org.opencontainers.image.source")}
    conts.sort(key=lambda x: x["nombre"])
    return {"disponible": True, "motor": motor, "contenedores": conts, "imagenes": imgs}


def _os_release():
    d = {}
    for line in leer("/etc/os-release").splitlines():
        if "=" in line:
            k, v = line.split("=", 1)
            d[k.strip()] = v.strip().strip('"')
    return {"id": d.get("ID"), "version": d.get("VERSION_ID"),
            "codigo": d.get("VERSION_CODENAME") or d.get("UBUNTU_CODENAME"),
            "nombre": d.get("PRETTY_NAME")}


# "openssl/jammy-updates,jammy-security 3.0.2-0ubuntu1.18 amd64 [upgradable from: 3.0.2-0ubuntu1.15]"
_RE_APT = re.compile(r"^(\S+?)/(\S+)\s+(\S+)\s+\S+\s+\[upgradable from: ([^\]]+)\]")


def en_fases():
    """Paquetes que Ubuntu da por fases (phased updates): «apt list --upgradable» los lista, pero
    «apt-get upgrade» no los instala hasta que le toca a la máquina (Fujitsu 2026-10-15: apparmor
    seguía «pendiente» tras Actualizar). Se leen de la simulación, que no toca nada."""
    rc, s, _ = run(["apt-get", "-s", "upgrade"], timeout=60)
    fases, dentro = set(), False
    for line in (s if rc == 0 else "").splitlines():
        if line.startswith("The following upgrades have been deferred due to phasing"):
            dentro = True
        elif dentro and line.startswith(" "):
            fases.update(line.split())
        else:
            dentro = False
    return fases


def paquetes_pendientes():
    if not which("apt"):
        return {"disponible": False}
    rc, s, err = run(["apt", "list", "--upgradable"], timeout=60)
    if rc != 0:
        return {"disponible": True, "error": (err or "rc=%d" % rc).strip()[-200:]}
    fases = en_fases()
    pend, retenidos = [], []
    for line in s.splitlines():
        m = _RE_APT.match(line.strip())
        if m and m.group(1) in fases:
            retenidos.append(m.group(1))
        elif m:
            pend.append({"paquete": m.group(1), "nueva": m.group(3), "instalada": m.group(4),
                         "seguridad": "-security" in m.group(2)})
    listas = [p for p in glob.glob("/var/lib/apt/lists/*") if p.endswith(("Release", "InRelease"))]
    ult = max((os.path.getmtime(p) for p in listas), default=None)
    return {"disponible": True, "total": len(pend),
            "seguridad": sum(1 for p in pend if p["seguridad"]),
            "listas_ts": int(ult) if ult else None,
            "paquetes": pend[:200], "fases": retenidos[:200]}


def actualizaciones():
    return {"so": _os_release(), "kernel": os.uname().release,
            "docker": imagenes_docker(), "apt": paquetes_pendientes()}


# ------------------------------------------------------------------------ red
def _gateway():
    for line in leer("/proc/net/route").splitlines()[1:]:
        p = line.split()
        if len(p) >= 3 and p[1] == "00000000":
            gw = int(p[2], 16)
            return p[0], "%d.%d.%d.%d" % (gw & 0xff, (gw >> 8) & 0xff, (gw >> 16) & 0xff, gw >> 24)
    return None, None


def red():
    ifaces = []
    for path in sorted(glob.glob("/sys/class/net/*")):
        name = os.path.basename(path)
        if name == "lo" or name.startswith(("veth", "br-", "docker", "virbr")):
            continue
        st = leer(os.path.join(path, "operstate")).strip()
        try:
            speed = int(leer(os.path.join(path, "speed")).strip())
        except ValueError:
            speed = None
        stats = {}
        for k in ("rx_bytes", "tx_bytes", "rx_errors", "tx_errors", "rx_dropped"):
            try:
                stats[k] = int(leer(os.path.join(path, "statistics", k)).strip())
            except ValueError:
                stats[k] = None
        ifaces.append({"nombre": name, "estado": st, "velocidad_mbps": speed if speed and speed > 0 else None,
                       **stats})
    iface_gw, gw = _gateway()
    r = {"interfaces": ifaces, "gateway": gw, "iface_gateway": iface_gw}
    if gw:
        rc, _, _ = run(["ping", "-c", "1", "-W", "2", gw], timeout=6)
        r["gateway_ok"] = rc == 0
    t0 = time.time()
    v, err = con_timeout(lambda: socket.getaddrinfo("github.com", 443)[0][4][0], 5)
    r["dns_ok"] = v is not None
    r["dns_ms"] = int((time.time() - t0) * 1000) if v else None
    r["dns_error"] = err
    t0 = time.time()
    try:
        s = socket.create_connection(("1.1.1.1", 443), timeout=3)
        s.close()
        r["internet_ok"] = True
        r["internet_ms"] = int((time.time() - t0) * 1000)
    except OSError as e:
        r["internet_ok"] = False
        r["internet_error"] = str(e)
    return r


# --------------------------------------------------------- extras (Fase 3)
def _arg(argv, nombre):
    """Valor de --nombre=valor o --nombre valor; None si no está."""
    for i, a in enumerate(argv):
        if a.startswith("--" + nombre + "="):
            return a.split("=", 1)[1]
        if a == "--" + nombre and i + 1 < len(argv) and not argv[i + 1].startswith("--"):
            return argv[i + 1]
    return None


def pings(objetivos):
    """{ip: {ms, perdida_pct}} con 5 ecos en paralelo (v4; antes 3: el switch
    limita ICMP y con 3 un eco perdido ya era el 33 %)."""
    res = {}
    lock = threading.Lock()

    def _uno(ip):
        rc, s, _ = run(["ping", "-c", "5", "-W", "1", "-i", "0.3", "-q", ip], timeout=12)
        r = {"ok": False, "ms": None, "perdida_pct": 100}
        m = re.search(r"([\d.]+)% packet loss", s)   # iputils viejo: "33.3333%"
        if m:
            r["perdida_pct"] = int(float(m.group(1)))
        m = re.search(r"= [\d.]+/([\d.]+)/", s)
        if m:
            r["ms"] = round(float(m.group(1)), 1)
        r["ok"] = rc == 0 and r["perdida_pct"] < 100
        with lock:
            res[ip] = r

    hilos = [threading.Thread(target=_uno, args=(ip,), daemon=True) for ip in objetivos if ip]
    for h in hilos:
        h.start()
    for h in hilos:
        h.join(12)
    return res


def wireguard():
    """Interfaces y pares de WireGuard (wg del host o dentro del contenedor wg-easy)."""
    rc, s, err = run(["wg", "show", "all", "dump"], timeout=10)
    if rc != 0:
        rc, s, err = run(["docker", "exec", "wg-easy", "wg", "show", "all", "dump"], timeout=15)
    if rc != 0:
        return {"ok": False, "error": (err or "sin wg").strip()[:120]}
    ifaces = {}
    ahora = int(time.time())
    for line in s.splitlines():
        p = line.split("\t")
        if len(p) == 5:
            ifaces[p[0]] = {"puerto": p[3], "pares": 0, "activos": 0}
        elif len(p) >= 9:
            it = ifaces.setdefault(p[0], {"puerto": None, "pares": 0, "activos": 0})
            it["pares"] += 1
            try:
                if int(p[5]) and ahora - int(p[5]) < 180:
                    it["activos"] += 1
            except ValueError:
                pass
    return {"ok": bool(ifaces), "interfaces": ifaces}


def nfs_exportaciones():
    rc, s, _ = run(["exportfs", "-s"], timeout=10)
    if rc != 0:
        return {"ok": False, "n": 0}
    return {"ok": True, "n": len([x for x in s.splitlines() if x.strip()])}


def certificado(sni, host="127.0.0.1", puerto=443):
    """Días que le quedan al certificado que sirve Caddy para ese nombre."""
    import ssl
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    try:
        with socket.create_connection((host, puerto), timeout=5) as sock:
            with ctx.wrap_socket(sock, server_hostname=sni) as ss:
                der = ss.getpeercert(binary_form=True)
    except (OSError, ssl.SSLError) as e:
        return {"ok": False, "error": str(e)[:120]}
    # sin verificación getpeercert() no devuelve dict: notAfter con openssl
    try:
        p = subprocess.run(["openssl", "x509", "-noout", "-enddate", "-issuer", "-inform", "DER"],
                           input=der, capture_output=True, timeout=10)
        out = p.stdout.decode("utf-8", "replace")
    except (OSError, subprocess.TimeoutExpired) as e:
        return {"ok": False, "error": str(e)}
    fin = emisor = ""
    for line in out.splitlines():
        if line.startswith("notAfter="):
            fin = line.split("=", 1)[1].strip()
        elif line.startswith("issuer="):
            emisor = line.split("=", 1)[1].strip()
    try:
        import datetime
        exp = datetime.datetime.strptime(fin, "%b %d %H:%M:%S %Y %Z")
        dias = (exp - datetime.datetime.utcnow()).days
    except ValueError:
        return {"ok": False, "error": "fecha ilegible: " + fin}
    return {"ok": True, "dias": dias, "caduca": fin, "emisor": emisor[:80], "sni": sni}


def backup_estado(ruta):
    """Último trabajo OK de Backup General (datos/state.json): {trabajos: {id: {last_ok, last_run}}}."""
    try:
        with open(ruta) as f:
            d = json.load(f)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": str(e)[:120]}
    jobs = d.get("jobs") or {}
    out = {}
    for jid, j in jobs.items():
        if not isinstance(j, dict) or jid.startswith("pruebas"):
            continue
        out[jid] = {"last_ok": j.get("last_ok"), "last_run": j.get("last_run")}
    corriendo = [r.get("job_name") for r in (d.get("runs") or []) if r.get("status") == "running"]
    return {"ok": True, "trabajos": out, "en_curso": corriendo}


def ip_publica(dominio):
    import urllib.request
    r = {"ok": False}
    try:
        with urllib.request.urlopen("https://api.ipify.org", timeout=6) as resp:
            r["ip"] = resp.read().decode().strip()
            r["ok"] = True
    except Exception as e:  # noqa: BLE001
        r["error"] = str(e)[:120]
    if dominio:
        v, err = con_timeout(lambda: socket.getaddrinfo(dominio, 443)[0][4][0], 5)
        r["dominio"] = dominio
        r["ip_dominio"] = v
    return r


def extras(argv):
    out = {}
    obj = _arg(argv, "ping")
    if obj:
        out["pings"] = pings([x.strip() for x in obj.split(",") if x.strip()])
    if "--wireguard" in argv:
        out["wireguard"] = wireguard()
    if "--nfs" in argv:
        out["nfs"] = nfs_exportaciones()
    sni = _arg(argv, "cert")
    if sni:
        out["cert"] = certificado(sni)
    ruta = _arg(argv, "backup-state")
    if ruta:
        out["backup"] = backup_estado(ruta)
    dom = _arg(argv, "ip-publica")
    if dom is not None or "--ip-publica" in argv:
        out["ip_publica"] = ip_publica(dom or "")
    return out


# ------------------------------------------------- firma de paquetes (v6)
# Copia LITERAL de ~/firma/firma.py (sin su docstring): la sonda es un solo
# fichero y tiene que comprobar las firmas en la máquina sin nada instalado.
# pruebas/fase8_descargar.py comprueba que no se desvía de la de referencia.
# >>> firma.py
import base64
import hashlib
import struct

NAMESPACE = "spufy-paquetes"
MAGIC = b"SSHSIG"

# ------------------------------------------------------------ Ed25519 (RFC 8032)
_p = 2 ** 255 - 19
_L = 2 ** 252 + 27742317777372353535851937790883648493


def _inv(x):
    return pow(x, _p - 2, _p)


_d = -121665 * _inv(121666) % _p
_SQRT_M1 = pow(2, (_p - 1) // 4, _p)


def _recover_x(y, sign):
    if y >= _p:
        return None
    x2 = (y * y - 1) * _inv(_d * y * y + 1)
    if x2 == 0:
        return None if sign else 0
    x = pow(x2, (_p + 3) // 8, _p)
    if (x * x - x2) % _p != 0:
        x = x * _SQRT_M1 % _p
    if (x * x - x2) % _p != 0:
        return None
    if (x & 1) != sign:
        x = _p - x
    return x


_gy = 4 * _inv(5) % _p
_gx = _recover_x(_gy, 0)
_G = (_gx, _gy, 1, _gx * _gy % _p)
_NEUTRO = (0, 1, 1, 0)


def _add(P, Q):
    A = (P[1] - P[0]) * (Q[1] - Q[0]) % _p
    B = (P[1] + P[0]) * (Q[1] + Q[0]) % _p
    C = 2 * P[3] * Q[3] * _d % _p
    D = 2 * P[2] * Q[2] % _p
    E, F, G, H = B - A, D - C, D + C, B + A
    return (E * F, G * H, F * G, E * H)


def _mul(s, P):
    Q = _NEUTRO
    while s > 0:
        if s & 1:
            Q = _add(Q, P)
        P = _add(P, P)
        s >>= 1
    return Q


def _igual(P, Q):
    return ((P[0] * Q[2] - Q[0] * P[2]) % _p == 0
            and (P[1] * Q[2] - Q[1] * P[2]) % _p == 0)


def _descomprimir(s):
    if len(s) != 32:
        return None
    y = int.from_bytes(s, "little")
    sign = y >> 255
    y &= (1 << 255) - 1
    x = _recover_x(y, sign)
    if x is None:
        return None
    return (x, y, 1, x * y % _p)


def ed25519_verificar(publica, mensaje, firma):
    """publica 32 bytes, firma 64 bytes -> bool."""
    if len(publica) != 32 or len(firma) != 64:
        return False
    A = _descomprimir(publica)
    R = _descomprimir(firma[:32])
    if A is None or R is None:
        return False
    s = int.from_bytes(firma[32:], "little")
    if s >= _L:
        return False
    h = int.from_bytes(hashlib.sha512(firma[:32] + publica + mensaje).digest(), "little") % _L
    return _igual(_mul(s, _G), _add(R, _mul(h, A)))


# --------------------------------------------------------- formato de OpenSSH
class _Lector:
    def __init__(self, b):
        self.b, self.i = b, 0

    def bytes(self, n):
        if self.i + n > len(self.b):
            raise ValueError("firma truncada")
        v = self.b[self.i:self.i + n]
        self.i += n
        return v

    def u32(self):
        return struct.unpack(">I", self.bytes(4))[0]

    def cadena(self):
        return self.bytes(self.u32())

    def fin(self):
        return self.i == len(self.b)


def _cadena(b):
    return struct.pack(">I", len(b)) + b


def clave_publica(linea):
    """«ssh-ed25519 AAAA… comentario» -> (blob de la clave, 32 bytes de la clave)."""
    partes = (linea or "").strip().split()
    if len(partes) < 2 or partes[0] != "ssh-ed25519":
        raise ValueError("no es una clave pública ssh-ed25519")
    blob = base64.b64decode(partes[1], validate=True)
    r = _Lector(blob)
    if r.cadena() != b"ssh-ed25519":
        raise ValueError("clave pública mal formada")
    k = r.cadena()
    if len(k) != 32 or not r.fin():
        raise ValueError("clave pública mal formada")
    return blob, k


def huella(linea):
    """SHA256:… igual que `ssh-keygen -lf`, para comprobarla a ojo."""
    blob, _ = clave_publica(linea)
    return "SHA256:" + base64.b64encode(hashlib.sha256(blob).digest()).decode().rstrip("=")


def _desarmar(texto):
    lineas = [x.strip() for x in texto.strip().splitlines()]
    if not lineas or lineas[0] != "-----BEGIN SSH SIGNATURE-----" or lineas[-1] != "-----END SSH SIGNATURE-----":
        raise ValueError("no es una firma SSH")
    return base64.b64decode("".join(lineas[1:-1]), validate=True)


_HASHES = {b"sha512": hashlib.sha512, b"sha256": hashlib.sha256}


def _comprobar(firma_texto, clave_linea, namespace, resumen_de):
    """resumen_de(nombre_hash) -> resumen del mensaje. Devuelve (ok, motivo)."""
    try:
        blob_conf, k_conf = clave_publica(clave_linea)
        r = _Lector(_desarmar(firma_texto))
        if r.bytes(6) != MAGIC or r.u32() != 1:
            return False, "firma con formato desconocido"
        blob_firma = r.cadena()
        ns = r.cadena()
        reservado = r.cadena()
        alg = r.cadena()
        sig = _Lector(r.cadena())
        if not r.fin():
            return False, "firma con datos de más"
        if blob_firma != blob_conf:
            return False, "firmado con otra clave"
        if ns != namespace.encode():
            return False, "firma para otro uso (%s)" % ns.decode("utf-8", "replace")
        if alg not in _HASHES:
            return False, "algoritmo de resumen no admitido"
        if sig.cadena() != b"ssh-ed25519":
            return False, "tipo de firma no admitido"
        s = sig.cadena()
        if not sig.fin():
            return False, "firma mal formada"
        firmado = MAGIC + _cadena(ns) + _cadena(reservado) + _cadena(alg) + _cadena(resumen_de(alg))
        if not ed25519_verificar(k_conf, firmado, s):
            return False, "la firma no corresponde al contenido"
        return True, "firma correcta"
    except (ValueError, struct.error) as e:
        return False, "firma ilegible: %s" % e


def verificar_bytes(datos, firma_texto, clave_linea, namespace=NAMESPACE):
    return _comprobar(firma_texto, clave_linea, namespace,
                      lambda alg: _HASHES[alg](datos).digest())


def verificar_fichero(ruta, ruta_firma, clave_linea, namespace=NAMESPACE):
    try:
        with open(ruta_firma, "r", encoding="ascii") as f:
            firma_texto = f.read(8192)
    except (OSError, UnicodeDecodeError) as e:
        return False, "no se pudo leer la firma: %s" % e

    def resumen(alg):
        h = _HASHES[alg]()
        with open(ruta, "rb") as f:
            for trozo in iter(lambda: f.read(1 << 20), b""):
                h.update(trozo)
        return h.digest()

    try:
        return _comprobar(firma_texto, clave_linea, namespace, resumen)
    except OSError as e:
        return False, "no se pudo leer el fichero: %s" % e
# <<< firma.py


# ------------------------------------------------------------ acciones (0.9)
# Lista CERRADA: la clave del panel solo puede ejecutar esta sonda, y la sonda
# solo hace esto. Cada acción valida sus argumentos con una expresión regular.
DISCO_RE = re.compile(r"^(sd[a-z]{1,3}|nvme\d+n\d+|hd[a-z])$")


def accion_smart_test(disco, tipo):
    if not DISCO_RE.match(disco or "") or tipo not in ("short", "long"):
        return {"ok": False, "error": "argumentos no válidos"}
    dev = "/dev/" + disco
    if not os.path.exists(dev):
        return {"ok": False, "error": "no existe " + dev}
    rc, s, err = run(["smartctl", "-t", tipo, dev], timeout=30)
    ok = rc == 0 or "has begun" in s
    m = re.search(r"Please wait (\d+) (minutes|hours)", s)
    dur = ("%s %s" % (m.group(1), "min" if m.group(2) == "minutes" else "h")) if m else "?"
    return {"ok": ok, "disco": dev, "tipo": tipo, "duracion": dur,
            "salida": (s if ok else (s + err)).strip()[-400:]}


def accion_smart_abort(disco):
    if not DISCO_RE.match(disco or ""):
        return {"ok": False, "error": "argumentos no válidos"}
    rc, s, err = run(["smartctl", "-X", "/dev/" + disco], timeout=30)
    return {"ok": rc == 0, "disco": "/dev/" + disco, "salida": (s + err).strip()[-300:]}


# Descarga de una versión nueva de un paquete NUESTRO (v6). Acción cerrada: el
# panel solo dice el proyecto; la web es fija (CG_WEB solo existe en desarrollo:
# por la clave forzada no llega el entorno del cliente), el nombre del fichero y
# la carpeta los deduce la sonda, y nada se queda si la firma no es buena.
WEB_VERSIONES = "https://web.spufy.org"
CLAVE_FIRMA = ("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAINAgGDthf6LHIpRai1ufrtq2Ub88YWZuAy9msqz47YkO "
               "spufy-paquetes")
AGENTE = "spufy-actualizador/cg-sonda-%d" % SONDA_VERSION
SLUG_RE = re.compile(r"^[a-z0-9]{2,40}$")
VERSION_RE = re.compile(r"^\d{1,4}\.\d{1,4}$")
PAQUETE_MAX = 2 * 1024 ** 3          # 2 GB: ningún paquete nuestro se acerca
LIBRE_MIN_PCT = 5                     # como el 95 % de los discos de la casa
ARQ = {"x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64"}


def _ver(v):
    return tuple(int(x) for x in v.split("."))


def _web():
    return (os.environ.get("CG_WEB") or WEB_VERSIONES).rstrip("/")


def _web_get(ruta, limite, destino=None):
    """Baja WEB+ruta; a memoria (bytes) o a `destino`. Corta si pasa de `limite`."""
    import ssl
    import urllib.request
    url = _web() + ruta
    req = urllib.request.Request(url, headers={"User-Agent": AGENTE})
    ctx = ssl.create_default_context() if url.startswith("https://") else None
    total, trozos = 0, []
    f = open(destino, "wb") if destino else None
    try:
        with urllib.request.urlopen(req, timeout=60, context=ctx) as r:
            while True:
                b = r.read(1 << 20)
                if not b:
                    break
                total += len(b)
                if total > limite:
                    raise ValueError("%s: más grande de lo anunciado" % ruta)
                if f:
                    f.write(b)
                else:
                    trozos.append(b)
    finally:
        if f:
            f.close()
    return total if destino else b"".join(trozos)


def _propios(slug):
    """Versión instalada (la más alta de las imágenes <slug>:X.Y en uso) y carpetas
    de compose de esos contenedores."""
    lista = _docker_get("/containers/json?all=1")
    instalada, dirs = None, set()
    for c in lista:
        try:
            tags = _docker_get("/images/%s/json" % c.get("ImageID")).get("RepoTags") or []
        except Exception:  # noqa: BLE001
            tags = []
        tags = tags + [c.get("Image") or ""]
        vers = [t.split(":", 1)[1] for t in tags if t.startswith(slug + ":") and VERSION_RE.match(t.split(":", 1)[1])]
        if not vers:
            continue
        v = max(vers, key=_ver)
        if instalada is None or _ver(v) > _ver(instalada):
            instalada = v
        d = (c.get("Labels") or {}).get("com.docker.compose.project.working_dir")
        if d:
            dirs.add(d)
    return instalada, sorted(dirs)


def _carpeta_actualizaciones(d):
    """<compose>/actualizaciones si existe, es carpeta de verdad y no sale de <compose>."""
    if not d or not os.path.isabs(d) or not os.path.isdir(d) or os.path.islink(d):
        return None
    a = os.path.join(d, "actualizaciones")
    if os.path.islink(a) or not os.path.isdir(a):
        return None
    if os.path.dirname(os.path.realpath(a)) != os.path.realpath(d):
        return None
    return a


def accion_descargar(slug):
    if not SLUG_RE.match(slug or ""):
        return {"ok": False, "error": "proyecto no válido"}
    arq = ARQ.get(os.uname().machine)
    if not arq:
        return {"ok": False, "error": "arquitectura sin paquetes: %s" % os.uname().machine}
    indice = _web_get("/versiones.json", 1 << 20)
    firma_indice = _web_get("/versiones.json.sig", 8192).decode("ascii", "replace")
    ok, motivo = verificar_bytes(indice, firma_indice, CLAVE_FIRMA)
    if not ok:
        return {"ok": False, "firma_mala": True, "error": "versiones.json: " + motivo}
    p = (json.loads(indice.decode("utf-8")).get("proyectos") or {}).get(slug)
    if not isinstance(p, dict) or not VERSION_RE.match(str(p.get("version", ""))):
        return {"ok": False, "error": "%s no está publicado en la web" % slug}
    nueva = p["version"]
    x = (p.get("paquetes") or {}).get(arq)
    if not isinstance(x, dict):
        return {"ok": False, "error": "%s %s no está publicado para %s" % (slug, nueva, arq)}
    try:
        tam = int(x.get("bytes"))
    except (TypeError, ValueError):
        return {"ok": False, "error": "tamaño no válido en versiones.json"}
    if not 0 < tam <= PAQUETE_MAX:
        return {"ok": False, "error": "tamaño fuera de límites: %s" % tam}
    instalada, dirs = _propios(slug)
    if instalada is None:
        return {"ok": False, "error": "no hay ningún contenedor de %s en esta máquina" % slug}
    r = {"ok": True, "proyecto": slug, "instalada": instalada, "nueva": nueva, "arq": arq}
    if _ver(nueva) <= _ver(instalada):
        r["al_dia"] = True
        return r
    destinos = [a for a in (_carpeta_actualizaciones(d) for d in dirs) if a]
    if not destinos:
        return dict(r, ok=False, error="sin carpeta actualizaciones/ junto al compose (%s)" % ", ".join(dirs or ["?"]))
    # el nombre y la ruta los pone la sonda, no el índice
    fichero = "%s-%s-%s.tar.gz" % (slug, nueva, arq)
    ruta_web = "/descargas/%s/%s" % (slug, fichero)
    viejo_re = re.compile(r"^%s-\d+\.\d+(-(amd64|arm64))?\.(tar\.gz|zip)(\.sig)?(\.parte)?$" % re.escape(slug))
    r["descargado"], r["borrados"] = [], []
    for a in destinos:
        final = os.path.join(a, fichero)
        if os.path.isfile(final) and verificar_fichero(final, final + ".sig", CLAVE_FIRMA)[0]:
            r["descargado"].append(final)
            r["ya_estaba"] = True
            continue
        st = os.statvfs(a)
        libre, total = st.f_bavail * st.f_frsize, st.f_blocks * st.f_frsize
        if libre - tam < total * LIBRE_MIN_PCT / 100:
            return dict(r, ok=False, error="sin sitio en %s (quedaría menos del %d %% libre)" % (a, LIBRE_MIN_PCT))
        parte = os.path.join(a, "." + fichero + ".parte")
        try:
            n = _web_get(ruta_web, tam, parte)
            if n != tam:
                raise ValueError("descargado %d de %d bytes" % (n, tam))
            sig = _web_get(ruta_web + ".sig", 8192)
            with open(parte + ".sig", "wb") as f:
                f.write(sig)
            ok, motivo = verificar_fichero(parte, parte + ".sig", CLAVE_FIRMA)
            if not ok:
                return dict(r, ok=False, firma_mala=True, error="%s: %s; NO se ha guardado" % (fichero, motivo))
            dueno = os.stat(a)
            for src, dst in ((parte + ".sig", final + ".sig"), (parte, final)):
                os.chown(src, dueno.st_uid, dueno.st_gid)
                os.chmod(src, 0o644)
                os.replace(src, dst)
            r["descargado"].append(final)
        finally:
            for t in (parte, parte + ".sig"):
                if os.path.exists(t):
                    os.remove(t)
        # en actualizaciones/ solo el último paquete de este proyecto
        for n in os.listdir(a):
            v = os.path.join(a, n)
            if viejo_re.match(n) and n not in (fichero, fichero + ".sig") \
                    and os.path.isfile(v) and not os.path.islink(v):
                os.remove(v)
                r["borrados"].append(v)
    return r


# Actualizar desde el panel (v6, Fase B). SOLO al pulsar el usuario y con
# «actualizar_desde_panel» encendido en Ajustes (apagado de serie). Acciones cerradas:
#   actualizar_contenedor:<nombre>  pull + up -d de SU servicio de compose; si el
#       contenedor nuevo no queda en marcha (y healthy, si tiene healthcheck), se
#       vuelve a la imagen anterior. Nunca el propio Control General.
#   apt_actualizar  apt-get update + upgrade (nunca dist-upgrade, autoremove ni
#       reinicio) en una unidad de systemd aparte: sigue aunque se corte el SSH o se
#       reinicie Docker. apt_estado lee cómo va.
NOMBRE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")
PROYECTO_PANEL = "controlgeneral"
ESPERA_ARRANQUE = 120


def _contenedor(nombre):
    for c in _docker_get("/containers/json?all=1"):
        if nombre in [(n or "").lstrip("/") for n in c.get("Names") or []]:
            return c
    return None


def _en_maquina(ruta, wd):
    """La ruta de un fichero de compose tal como se ve EN la máquina. Los paneles que se actualizan solos
    (Ajustes → Actualización) levantan su compose desde dentro de un contenedor, y las etiquetas quedan con
    SUS rutas (/host/home/RenombrarMusica/docker-compose.yml, /project/.env…), que en la máquina no existen
    (boira, 2026-10-11: «open /host/home/RenombrarMusica/docker-compose.yml: no such file or directory»).
    Si no existe, el fichero del mismo nombre en la carpeta del proyecto (working_dir sí es la de la máquina)."""
    if os.path.isfile(ruta):
        return ruta
    otra = os.path.join(wd, os.path.basename(ruta))
    return otra if os.path.isfile(otra) else ruta


def _compose_cmd(etq):
    """docker compose con los ficheros y el proyecto del contenedor (como lo levantó el usuario)."""
    wd = etq.get("com.docker.compose.project.working_dir")
    ficheros = [f for f in (etq.get("com.docker.compose.project.config_files") or "").split(",") if f]
    cmd = ["docker", "compose", "--project-directory", wd, "-p", etq.get("com.docker.compose.project")]
    for f in ficheros:
        cmd += ["-f", _en_maquina(f, wd)]
    env = etq.get("com.docker.compose.project.environment_file")
    if env and os.path.isfile(_en_maquina(env, wd)):
        cmd += ["--env-file", _en_maquina(env, wd)]
    return cmd


def _arrancado(nombre, segundos):
    """Espera a que el contenedor esté running (y healthy si tiene healthcheck)."""
    fin = time.time() + segundos
    estable = 0
    while time.time() < fin:
        c = _contenedor(nombre)
        st = (c or {}).get("Status", "").lower()
        if c and c.get("State") == "running":
            if "(healthy)" in st:
                return True, st
            if "health: starting" not in st and "(unhealthy)" not in st:
                estable += 1
                if estable >= 4:      # ~20 s seguidos en marcha sin healthcheck
                    return True, st
        else:
            estable = 0
        time.sleep(5)
    return False, ((c or {}).get("Status") or "no existe")


def accion_actualizar_contenedor(nombre):
    if not NOMBRE_RE.match(nombre or ""):
        return {"ok": False, "error": "nombre no válido"}
    if not which("docker"):
        return {"ok": False, "error": "no está el cliente docker en la máquina"}
    c = _contenedor(nombre)
    if not c:
        return {"ok": False, "error": "no existe el contenedor %s" % nombre}
    etq = c.get("Labels") or {}
    servicio = etq.get("com.docker.compose.service")
    wd = etq.get("com.docker.compose.project.working_dir")
    if not servicio or not wd or not etq.get("com.docker.compose.project"):
        return {"ok": False, "error": "%s no se creó con docker compose: actualizarlo a mano" % nombre}
    if etq.get("com.docker.compose.project") == PROYECTO_PANEL:
        return {"ok": False, "error": "el propio Control General se actualiza con su paquete"}
    if not os.path.isdir(wd):
        return {"ok": False, "error": "no existe la carpeta del compose: %s" % wd}
    ref, viejo = c.get("Image"), c.get("ImageID")
    if not ref or ref.startswith("sha256:") or "@sha256:" in ref:
        return {"ok": False, "error": "imagen fijada por digest o sin nombre: no se toca"}
    base = _compose_cmd(etq)
    r = {"contenedor": nombre, "servicio": servicio, "imagen": ref, "anterior": viejo}
    rc, out, err = run(base + ["pull", servicio], timeout=900)
    if rc != 0:
        return dict(r, ok=False, error="no se pudo bajar la imagen: " + (err or out).strip()[-300:])
    try:
        nuevo = _docker_get("/images/%s/json" % ref).get("Id")
    except Exception as e:  # noqa: BLE001
        return dict(r, ok=False, error="imagen bajada pero ilegible: %s" % e)
    r["nueva"] = nuevo
    if nuevo == viejo:
        return dict(r, ok=True, al_dia=True)
    rc, out, err = run(base + ["up", "-d", "--no-deps", servicio], timeout=300)
    bien, estado = (False, (err or out).strip()[-300:]) if rc != 0 else _arrancado(nombre, ESPERA_ARRANQUE)
    if bien:
        return dict(r, ok=True, estado=estado)
    # vuelta atrás: la etiqueta vuelve a la imagen anterior y se recrea con ella
    rc_l, logs, _ = run(["docker", "logs", "--tail", "15", nombre], timeout=20)
    run(["docker", "tag", viejo, ref], timeout=30)
    rc2, out2, err2 = run(base + ["up", "-d", "--no-deps", "--pull", "never", servicio], timeout=300)
    atras, estado2 = _arrancado(nombre, ESPERA_ARRANQUE) if rc2 == 0 else (False, (err2 or out2).strip()[-200:])
    return dict(r, ok=False, vuelta_atras=atras,
                error="la imagen nueva no arrancó (%s); %s" % (estado, "se ha vuelto a la anterior y funciona"
                                                              if atras else "OJO: la vuelta atrás tampoco arrancó (%s)" % estado2),
                logs=(logs or "")[-1500:])


APT_UNIDAD = "cg-apt"
APT_LOG = "/var/log/cg-apt.log"


def accion_apt_actualizar():
    if not which("apt-get") or not which("systemd-run"):
        return {"ok": False, "error": "esta máquina no tiene apt-get o systemd-run"}
    rc, out, _ = run(["systemctl", "is-active", APT_UNIDAD], timeout=10)
    if out.strip() in ("active", "activating"):
        return {"ok": False, "error": "ya hay una actualización en marcha"}
    run(["systemctl", "reset-failed", APT_UNIDAD], timeout=10)
    guion = ("export DEBIAN_FRONTEND=noninteractive; date; apt-get update && "
             "apt-get -y -o Dpkg::Options::=--force-confdef -o Dpkg::Options::=--force-confold upgrade; "
             "rc=$?; date; echo \"FIN rc=$rc\"; exit $rc")
    rc, out, err = run(["systemd-run", "--unit", APT_UNIDAD, "--collect", "--property", "StandardOutput=file:" + APT_LOG,
                        "--property", "StandardError=file:" + APT_LOG, "/bin/sh", "-c", guion], timeout=30)
    if rc != 0:
        return {"ok": False, "error": (err or out).strip()[-300:]}
    return {"ok": True, "lanzado": True, "unidad": APT_UNIDAD}


def accion_apt_estado():
    rc, out, _ = run(["systemctl", "is-active", APT_UNIDAD], timeout=10)
    log = leer(APT_LOG)
    m = re.findall(r"FIN rc=(\d+)", log)
    activa = out.strip() in ("active", "activating")
    return {"ok": True, "en_marcha": activa, "terminado": bool(m) and not activa,
            "rc": int(m[-1]) if m else None, "log": log[-2500:],
            "reinicio": os.path.exists("/var/run/reboot-required") or bool(kernel_nuevo())}


# Auto-actualización (v7). El panel manda por stdin {"sonda": texto, "firma": .sig}
# de la sonda que trae su versión. Solo se acepta si: está firmada con la clave de
# la casa (el panel no puede colar otro programa), es de versión MAYOR, compila y,
# ejecutada aparte, responde con esa versión. Se guarda la anterior como .bak-vN.
SONDA_INSTALADA = "/usr/local/bin/cg-sonda"
_RE_SONDA_VERSION = re.compile(r"^SONDA_VERSION = (\d+)\s*$", re.M)


def accion_actualizar_sonda():
    if os.path.realpath(sys.argv[0] or "") != SONDA_INSTALADA:
        return {"ok": False, "error": "esta sonda no está instalada en %s (modo stdin): nada que actualizar" % SONDA_INSTALADA}
    try:
        datos = json.loads(sys.stdin.buffer.read(4 << 20).decode("utf-8"))
        texto, firma = datos["sonda"], datos["firma"]
        assert isinstance(texto, str) and isinstance(firma, str)
    except Exception:  # noqa: BLE001
        return {"ok": False, "error": "no llegó una sonda legible por stdin"}
    cuerpo = texto.encode("utf-8")
    ok, motivo = verificar_bytes(cuerpo, firma, CLAVE_FIRMA)
    if not ok:
        return {"ok": False, "firma_mala": True, "error": "sonda nueva rechazada: " + motivo}
    m = _RE_SONDA_VERSION.search(texto)
    nueva = int(m.group(1)) if m else 0
    if nueva <= SONDA_VERSION:
        return {"ok": True, "al_dia": True, "de": SONDA_VERSION, "a": nueva}
    try:
        compile(texto, "cg-sonda", "exec")
    except SyntaxError as e:
        return {"ok": False, "error": "la sonda nueva no compila con el python3 de aquí: %s" % e}
    tmp = SONDA_INSTALADA + ".nueva"
    with open(tmp, "wb") as f:
        f.write(cuerpo)
    os.chmod(tmp, 0o755)
    rc, out, err = run([tmp, "--sin-docker", "--sin-red"], timeout=90, sin=("SSH_ORIGINAL_COMMAND",))
    try:
        prueba = json.loads(out.strip().splitlines()[-1])
        bien = prueba.get("sonda_version") == nueva and "sistema" in prueba
    except (ValueError, IndexError, AttributeError):
        bien = False
    if not bien:
        os.remove(tmp)
        return {"ok": False, "error": "la sonda v%d no respondió bien al probarla; se queda la v%d (%s)"
                % (nueva, SONDA_VERSION, (err or out).strip()[-200:])}
    bak = "%s.bak-v%d" % (SONDA_INSTALADA, SONDA_VERSION)
    with open(SONDA_INSTALADA, "rb") as f, open(bak, "wb") as g:
        g.write(f.read())
    os.chmod(bak, 0o755)
    os.replace(tmp, SONDA_INSTALADA)
    return {"ok": True, "de": SONDA_VERSION, "a": nueva, "copia": bak}


ACCIONES = {"smart_test": (accion_smart_test, 2), "smart_abort": (accion_smart_abort, 1),
            "descargar": (accion_descargar, 1),
            "actualizar_contenedor": (accion_actualizar_contenedor, 1),
            "apt_actualizar": (accion_apt_actualizar, 0), "apt_estado": (accion_apt_estado, 0),
            "actualizar_sonda": (accion_actualizar_sonda, 0)}


def accion(texto):
    partes = (texto or "").split(":")
    nombre, args = partes[0], partes[1:]
    if nombre not in ACCIONES or len(args) != ACCIONES[nombre][1]:
        return {"ok": False, "error": "acción desconocida o argumentos incorrectos: %s" % texto}
    try:
        r = ACCIONES[nombre][0](*args)
    except Exception as e:  # noqa: BLE001
        r = {"ok": False, "error": "%s: %s" % (type(e).__name__, e)}
    r["accion"] = nombre
    r["sonda_version"] = SONDA_VERSION
    return r


# ----------------------------------------------------------------------- main
def main():
    argv = sys.argv[1:]
    orig = os.environ.get("SSH_ORIGINAL_COMMAND", "")
    # (v9) Si me ejecuta otra sonda para probarme antes de sustituirse (cg-sonda.nueva), esa
    # variable es la orden que le llegó a ELLA (actualizar_sonda), no a mí: respondo mis datos.
    if os.path.basename(sys.argv[0] or "").endswith(".nueva"):
        orig = ""
    if orig:
        argv += shlex.split(orig)[1:] if orig.split() and not orig.startswith("-") else shlex.split(orig)
    a = _arg(argv, "accion")
    if a:
        sys.stdout.write(json.dumps(accion(a), ensure_ascii=False) + "\n")
        return
    con_smart = "--smart" in argv
    t0 = time.time()
    cpu_ini = _cpu_sample()
    time.sleep(0.5)  # ventana para el % de CPU
    res = {"sonda_version": SONDA_VERSION, "generado": int(time.time()), "root": os.geteuid() == 0,
           "sistema": sistema(cpu_ini)}
    for clave, fn, arg in (("discos", discos, con_smart), ("mdadm", mdadm, None),
                           ("montajes", montajes, None),
                           ("docker", docker if "--sin-docker" not in argv else None, None),
                           ("red", red if "--sin-red" not in argv else None, None),
                           ("extras", extras, argv),
                           ("actualizaciones", actualizaciones if "--actualizaciones" in argv else None, None)):
        if fn is None:
            continue
        try:
            res[clave] = fn(arg) if arg is not None else fn()
        except Exception as e:  # noqa: BLE001
            res[clave] = {"error": "%s: %s" % (type(e).__name__, e)}
    res["con_smart"] = con_smart
    res["duracion_s"] = round(time.time() - t0, 2)
    sys.stdout.write(json.dumps(res, ensure_ascii=False))
    sys.stdout.write("\n")


if __name__ == "__main__":
    main()
