sábado, 26 de septiembre de 2026

Aviso automático por Telegram con la ubicación CDP/LLDP de tu Raspberry Pi


Si tienes un mini-servidor Linux (una Raspberry Pi, una WLAN Pi, cualquier placa ARM) desplegado en un sitio donde no vas a estar delante justo cuando lo enciendas, este componente resuelve una pregunta muy concreta sin que tengas que ir físicamente a comprobarlo: ¿ya tiene IP? ¿tiene Internet? ¿a qué puerto del switch lo han conectado?


Es un servicio en background, ligero (solo librería estándar de Python, sin dependencias externas), que:

  • Detecta el vecino de red vía CDP/LLDP: el switch se anuncia solo, sin que tengas que preguntarle nada.
  • Comprueba si hay Internet (por conexión TCP real, nunca ping, que muchos firewalls bloquean sin que eso signifique nada malo).
  • Manda un Telegram con el hostname, la IP y la MAC de cada interfaz, y el switch/puerto detectado, solo cuando aporta algo nuevo.

Al final del artículo tienes los tres scripts completos, listos para copiar.


🧩 1. Conceptos básicos

CDP / LLDP: protocolos con los que un switch (Cisco u otros compatibles) anuncia periódicamente, en cada puerto, quién es y a qué puerto exacto está conectado el equipo, sin que nadie tenga que preguntarlo. CDP es de Cisco; LLDP es el estándar abierto equivalente. En Linux, el demonio lldpd escucha los dos a la vez, sin ninguna configuración extra para recibir (solo la necesitarías para emitir, que aquí no hace falta).

systemd: el gestor de servicios en background de Linux. Un servicio se define en un fichero .service dentro de /etc/systemd/system/, y se controla con systemctl.

Bot de Telegram: una cuenta especial, gratuita y creable en cinco minutos, que permite a un programa mandar mensajes automáticos a un chat. Se identifica con un token (secreto) y manda a un chat_id concreto.

Todos los comandos de abajo, salvo que se diga lo contrario, se ejecutan por SSH en tu servidor Linux, no en tu propio ordenador.


⚙️ 2. Instalar lldpd (CDP/LLDP)

Ojo con dos problemas reales encontrados en un despliegue sobre Debian/Armbian algo desactualizado; si a ti no te aparecen, los comandos no hacen ningún daño de todas formas:

# si tu sources.list tiene una línea "bullseye-backports" muerta:
sudo sed -i 's|^deb http://deb.debian.org/debian bullseye-backports|#deb http://deb.debian.org/debian bullseye-backports|' /etc/apt/sources.list

sudo apt-get update
sudo apt-get install -y lldpd

Si el install falla con un 404 Not Found en una dependencia como libsnmp40, es un desfase temporal del mirror de seguridad de Debian. Fija la versión que sí está disponible en el repo principal:

apt-cache madison libsnmp40 libsnmp-base
# usa la versión listada bajo "deb.debian.org/debian bullseye/main"
sudo apt-get install -y lldpd libsnmp40=VERSION_DISPONIBLE libsnmp-base=VERSION_DISPONIBLE

Actívalo y espera un poco a que el switch anuncie (Cisco lo hace cada 60 segundos por defecto):

sudo systemctl enable --now lldpd
sleep 90 && sudo lldpcli show neighbors -f keyvalue

Comprobación: deberías ver líneas como lldp.eth0.chassis.name=SWITCH-EJEMPLO.tu-dominio.local y lldp.eth0.port.ifname=Gi0/4. Si sigue vacío con el cable conectado, revisa sudo lldpcli show configuration.


🤖 3. Crear el bot de Telegram

Esto se hace en la app de Telegram, no en una terminal:

  • Busca @BotFather y ábrele un chat.
  • Envíale /newbot, dale un nombre y un usuario (debe terminar en bot).
  • Te da un token (algo como 123456789:AAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx). Guárdalo, no lo compartas.
  • Busca tu bot por su usuario y mándale cualquier mensaje (un bot no puede escribirte primero).
  • Abre en el navegador https://api.telegram.org/bot TU_TOKEN /getUpdates (sin espacios) y busca "chat":{"id":: ese número es tu chat_id.

Si el último paso devuelve "result":[] vacío, es que aún no le has enviado ningún mensaje al bot: mándaselo y recarga.


📦 4. Subir los scripts y configurar

Copia los tres ficheros de más abajo tal cual a tu servidor (por ejemplo con scp, o pegándolos directamente por SSH con nano) en un mismo directorio, por ejemplo ~/notificador/:

  • estado_pi.py: la lógica, leer IPs/MACs, comprobar Internet, preguntar a lldpd, mandar el Telegram.
  • notificador_estado.py: el bucle que decide cuándo avisar.
  • notificador-estado.service: la unidad systemd.

Aviso de compatibilidad: si tu servidor tiene Python 3.9 o anterior, comprueba que ambos scripts .py tienen la línea from __future__ import annotations justo después del docstring inicial (ya la llevan en el código de abajo). Sin ella, Python 3.9 revienta con TypeError: unsupported operand type(s) for | al arrancar, porque el código usa anotaciones de tipo modernas (str | None) que solo se evalúan sin fallar desde Python 3.10. Compruébalo con python3 --version antes de darlo por hecho.

Mueve la unidad systemd a su sitio:

sudo cp ~/notificador/notificador-estado.service /etc/systemd/system/
cd ~/notificador && python3 -c "import estado_pi, notificador_estado; print('OK')"

Ese último comando no es opcional: comprueba de verdad que el código arranca en tu Python (no solo que está bien escrito). Debe imprimir OK sin ningún Traceback.

Crea el fichero de configuración (sustituye los placeholders por tus valores reales):

umask 077 && cat > ~/notificador/notificador_estado.conf <<'EOF'
TELEGRAM_BOT_TOKEN=TU_TOKEN_AQUI
TELEGRAM_CHAT_ID=TU_CHAT_ID_AQUI
INTERVALO_SEGUNDOS=30
INTERFAZ_WAN=eth0
EOF
chmod 600 ~/notificador/notificador_estado.conf

INTERFAZ_WAN depende de tu equipo: compruébala antes con ip -o -4 addr show (no asumas eth0 sin más).

Y arranca el servicio:

sudo systemctl daemon-reload
sudo systemctl enable --now notificador-estado.service
sudo journalctl -u notificador-estado.service --no-pager -n 30

Comprobación final: sin ningún Traceback en el log, y en tu chat de Telegram con el bot debe llegar el primer mensaje, con este aspecto:

Estado Raspberry Pi Wireless
Host: mi-servidor
2026-09-26 18:17:12

Internet: si
Interfaces:
  eth1: (sin IP)
  eth0: 10.0.0.50 [02:81:fd:60:32:15]
  wlan0: 169.254.13.15 [e0:e1:a9:36:57:d1]
  wlan1: (sin IP)
Vecino CDP/LLDP:
  switch: SWITCH-EJEMPLO.tu-dominio.local
  puerto: Gi0/4
  protocolo: LLDP

El servicio no manda un mensaje por cada ciclo: solo la primera vez que hay Internet tras arrancar, o cada vez que cambia la IP de INTERFAZ_WAN. Un cambio de vecino CDP aislado viaja dentro del próximo mensaje que sí se envíe por uno de esos dos motivos, en vez de generar uno propio.


🧾 5. Los tres scripts completos

estado_pi.py

"""Logica compartida entre notificador_estado.py (systemd) y la tool MCP
notificar_estado_telegram() de tshark_server.py: snapshot de IPs/Internet/CDP
y envio a Telegram. Solo libreria estandar (corre igual bajo el python3 de
sistema que bajo el venv mcp_env).
"""

from __future__ import annotations

import json
import re
import socket
import subprocess
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime

INTERFACES = ["eth1", "eth0", "wlan0", "wlan1"]
ENDPOINTS_INTERNET = [("1.1.1.1", 443), ("8.8.8.8", 443), ("9.9.9.9", 443)]


def cargar_config(ruta: str) -> dict:
    """Parsea el fichero de config CLAVE=valor (sin secciones, una por linea)."""
    config = {}
    with open(ruta) as f:
        for linea in f:
            linea = linea.strip()
            if not linea or linea.startswith("#") or "=" not in linea:
                continue
            clave, _, valor = linea.partition("=")
            config[clave.strip()] = valor.strip()
    return config


def obtener_ip_interfaz(interfaz: str) -> str | None:
    try:
        r = subprocess.run(["ip", "-4", "-o", "addr", "show", "dev", interfaz],
                            capture_output=True, text=True, timeout=5)
    except Exception:
        return None
    m = re.search(r"inet (\d+\.\d+\.\d+\.\d+)", r.stdout)
    return m.group(1) if m else None


def obtener_mac_interfaz(interfaz: str) -> str | None:
    """Lee la MAC directamente de /sys/class/net (sin subprocess, siempre
    disponible en Linux)."""
    try:
        with open(f"/sys/class/net/{interfaz}/address") as f:
            mac = f.read().strip()
    except OSError:
        return None
    return mac or None


def hay_internet(endpoints=ENDPOINTS_INTERNET, timeout: float = 3) -> bool:
    """TCP connect a IPs fiables (nunca ICMP: muchos firewalls lo bloquean).
    Usa la ruta que elija el kernel, no fuerza bind a ninguna interfaz."""
    for host, puerto in endpoints:
        try:
            with socket.create_connection((host, puerto), timeout=timeout):
                return True
        except OSError:
            continue
    return False


def obtener_vecino_cdp() -> dict | None:
    """Vecino CDP visto por lldpd en eth1, via 'lldpcli show neighbors -f keyvalue'.
    Ese formato produce lineas planas clave=valor, mas estable entre versiones de
    lldpd que el JSON anidado."""
    try:
        r = subprocess.run(["sudo", "-n", "lldpcli", "show", "neighbors", "-f", "keyvalue"],
                            capture_output=True, text=True, timeout=10)
    except Exception:
        return None
    salida = r.stdout.strip()
    if not salida:
        return None
    vecino = {}
    for linea in salida.splitlines():
        clave, _, valor = linea.partition("=")
        clave, valor = clave.strip(), valor.strip()
        if clave.endswith(".chassis.name"):
            vecino["switch"] = valor
        elif clave.endswith(".port.ifname") or clave.endswith(".port.descr"):
            vecino.setdefault("puerto", valor)
        elif clave.endswith(".via"):
            vecino["protocolo"] = valor
    return vecino or None


def construir_snapshot(config: dict) -> dict:
    return {
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "hostname": socket.gethostname(),
        "ips": {iface: obtener_ip_interfaz(iface) for iface in INTERFACES},
        "macs": {iface: obtener_mac_interfaz(iface) for iface in INTERFACES},
        "internet": hay_internet(),
        "cdp": obtener_vecino_cdp(),
    }


def formatear_mensaje_telegram(snapshot: dict) -> str:
    lineas = [
        "Estado Raspberry Pi Wireless",
        f"Host: {snapshot.get('hostname', '?')}",
        snapshot["timestamp"],
        "",
        f"Internet: {'si' if snapshot['internet'] else 'NO'}",
        "Interfaces:",
    ]
    for iface in INTERFACES:
        ip = snapshot["ips"].get(iface)
        mac = snapshot.get("macs", {}).get(iface)
        detalle = ip or "(sin IP)"
        if mac:
            detalle += f" [{mac}]"
        lineas.append(f"  {iface}: {detalle}")

    cdp = snapshot.get("cdp")
    lineas.append("Vecino CDP/LLDP:")
    if cdp:
        lineas.append(f"  switch: {cdp.get('switch', '?')}")
        lineas.append(f"  puerto: {cdp.get('puerto', '?')}")
        lineas.append(f"  protocolo: {cdp.get('protocolo', '?')}")
    else:
        lineas.append("  (sin vecino CDP detectado)")

    return "\n".join(lineas)


def enviar_telegram(token: str, chat_id: str, texto: str) -> bool:
    """POST a la API HTTP de Telegram via urllib puro (sin 'requests').
    Sin parse_mode: un nombre de switch/puerto con caracteres especiales
    rompería el parseo Markdown/HTML de Telegram y haría fallar el envio."""
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    datos = urllib.parse.urlencode({"chat_id": chat_id, "text": texto}).encode()
    try:
        req = urllib.request.Request(url, data=datos, method="POST")
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status == 200
    except (urllib.error.URLError, TimeoutError, OSError):
        return False


def cargar_json(ruta: str) -> dict | None:
    try:
        with open(ruta) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return None


def guardar_json(ruta: str, datos: dict) -> None:
    with open(ruta, "w") as f:
        json.dump(datos, f)

notificador_estado.py

"""Servicio en background (systemd, notificador-estado.service) que avisa por
Telegram del estado de la Raspberry Pi cuando aporta informacion nueva:

  - la primera vez que hay Internet tras arrancar el servicio (aunque la IP
    sea la misma que antes de reiniciar: es la senal de "ya estoy disponible"),
  - o cada vez que cambia la IP de la interfaz WAN (DHCP renovado con IP
    distinta, cambio de VLAN, etc.).

No notifica por flapping de Internet sin cambio de IP ni por un cambio de
vecino CDP aislado (ese dato solo se reporta dentro de un mensaje que ya se
envia por alguno de los dos motivos anteriores). Logica compartida con la
tool MCP notificar_estado_telegram() (tshark_server.py) vive en estado_pi.py.
"""

from __future__ import annotations

import os
import time
from datetime import datetime

import estado_pi

BASE_DIR = os.path.expanduser("~/mcp_tshark")
RUTA_CONF = os.path.join(BASE_DIR, "notificador_estado.conf")
RUTA_ESTADO = os.path.join(BASE_DIR, "notificador_estado_ultimo.json")


def cargar_ultima_ip_notificada(ruta: str) -> str | None:
    datos = estado_pi.cargar_json(ruta)
    return datos.get("ip_notificada") if datos else None


def guardar_ultima_ip_notificada(ruta: str, ip: str | None) -> None:
    estado_pi.guardar_json(ruta, {"ip_notificada": ip})


def bucle_principal() -> None:
    config = estado_pi.cargar_config(RUTA_CONF)
    intervalo = int(config.get("INTERVALO_SEGUNDOS", 30))
    interfaz_wan = config.get("INTERFAZ_WAN", "eth1")

    ip_notificada = cargar_ultima_ip_notificada(RUTA_ESTADO)
    notificado_tras_arranque = False  # se resetea en cada arranque del proceso

    while True:
        try:
            snapshot = estado_pi.construir_snapshot(config)
            ip_actual = snapshot["ips"].get(interfaz_wan)

            debe_notificar = (
                (not notificado_tras_arranque and snapshot["internet"])
                or (ip_actual is not None and ip_actual != ip_notificada)
            )

            if debe_notificar:
                texto = estado_pi.formatear_mensaje_telegram(snapshot)
                ok = estado_pi.enviar_telegram(
                    config["TELEGRAM_BOT_TOKEN"], config["TELEGRAM_CHAT_ID"], texto)
                if ok:
                    notificado_tras_arranque = True
                    ip_notificada = ip_actual
                    guardar_ultima_ip_notificada(RUTA_ESTADO, ip_actual)
                else:
                    print(f"[{datetime.now()}] Fallo enviando a Telegram, se reintentara")
        except Exception as e:
            print(f"[{datetime.now()}] Error en iteracion: {e}")

        time.sleep(intervalo)


if __name__ == "__main__":
    bucle_principal()

notificador-estado.service

[Unit]
Description=Notificador de estado (IP + Internet + vecino CDP) - Raspberry Pi Wireless
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=mcp.user
WorkingDirectory=/home/mcp.user/mcp_tshark
ExecStart=/usr/bin/python3 -u /home/mcp.user/mcp_tshark/notificador_estado.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

Recuerda: el TELEGRAM_BOT_TOKEN y el TELEGRAM_CHAT_ID van solo en tu fichero notificador_estado.conf (permisos 600, fuera de cualquier repositorio), nunca pegados en estos scripts.

No hay comentarios:

Publicar un comentario

Aviso automático por Telegram con la ubicación CDP/LLDP de tu Raspberry Pi

Si tienes un mini-servidor Linux (una Raspberry Pi, una WLAN Pi , cualquier placa ARM) desplegado en un sitio donde no vas a estar delante j...