#!/usr/bin/env python3 """ Vigilante de servicios — Ciberkit ================================= Comprueba que tus webs, APIs y servidores responden y avisa SOLO cuando algo cambia (se cae o se recupera), por consola y, si quieres, en Slack, Discord o Microsoft Teams. Pensado para ejecutarse con cron o el Programador de tareas, o en bucle con --cada. Qué comprueba cada objetivo: https://web/ruta Código HTTP, tiempo de respuesta, texto esperado y días que le quedan al certificado TLS. tcp:host:puerto Que el puerto acepta conexiones (SSH, base de datos…). dns:nombre Que el nombre se resuelve. Opciones por objetivo (en la misma línea, separadas por espacios): esperar=200 Código HTTP exacto (por defecto, cualquier 2xx o 3xx). contiene="texto" La respuesta debe incluir ese texto. lento=1500 Avisar si tarda más de esos milisegundos. nombre="API pagos" Nombre para los avisos. Con --global, cuando algo falla desde aquí se pregunta a nodos de varios países (check-host.net) para saber si está caído para todos o solo para ti. Úsalo SOLO con servicios propios o con autorización. Sin dependencias: Python 3.7+. Ejemplos -------- python vigilar_servicios.py https://miweb.com tcp:mi-servidor:22 python vigilar_servicios.py --lista servicios.txt --estado estado.json python vigilar_servicios.py --lista servicios.txt --cada 60 --webhook https://hooks.slack.com/services/... python vigilar_servicios.py https://miweb.com --global --csv historial.csv Ejemplo de servicios.txt: # Web principal: debe dar 200 y contener «Bienvenido» https://miweb.com esperar=200 contiene="Bienvenido" lento=2000 https://api.miweb.com/salud nombre="API" tcp:db.interna:5432 nombre="PostgreSQL" dns:miweb.com Código de salida: 0 todo bien, 1 algún aviso (lento, certificado a punto de caducar), 2 algo caído, 3 error de configuración. """ import argparse import csv import datetime import json import os import shlex import socket import ssl import sys import time import urllib.error import urllib.parse import urllib.request UA = "ciberkit-vigilante/1" CERT_AVISO_DIAS = 14 # ───────────────────────── Configuración ───────────────────────── def parse_objetivo(linea): """«https://x.com esperar=200 contiene="ok"» → dict. Lanza ValueError si no se entiende.""" partes = shlex.split(linea, comments=True) if not partes: return None destino, opciones = partes[0], {} for p in partes[1:]: if "=" not in p: raise ValueError("Opción «{}» no válida: usa clave=valor".format(p)) k, v = p.split("=", 1) opciones[k.strip().lower()] = v o = {"destino": destino, "nombre": opciones.get("nombre") or destino} if destino.startswith(("http://", "https://")): u = urllib.parse.urlsplit(destino) if not u.hostname: raise ValueError("URL sin servidor: {}".format(destino)) o["tipo"] = "http" elif destino.startswith("tcp:"): resto = destino[4:] host, _, puerto = resto.rpartition(":") if not host or not puerto.isdigit() or not 0 < int(puerto) < 65536: raise ValueError("Formato TCP: tcp:host:puerto (por ejemplo tcp:servidor:22)") o.update(tipo="tcp", host=host.strip("[]"), puerto=int(puerto)) elif destino.startswith("dns:"): if not destino[4:]: raise ValueError("Formato DNS: dns:nombre") o.update(tipo="dns", host=destino[4:]) else: raise ValueError("Objetivo «{}» no válido: usa https://…, tcp:host:puerto o dns:nombre".format(destino)) if "esperar" in opciones: if not opciones["esperar"].isdigit(): raise ValueError("esperar= debe ser un código HTTP") o["esperar"] = int(opciones["esperar"]) if "contiene" in opciones: o["contiene"] = opciones["contiene"] if "lento" in opciones: if not opciones["lento"].isdigit(): raise ValueError("lento= debe ser un número de milisegundos") o["lento"] = int(opciones["lento"]) return o def leer_lista(ruta): objetivos = [] with open(ruta, encoding="utf-8-sig") as f: for n, linea in enumerate(f, 1): try: o = parse_objetivo(linea) except ValueError as e: raise ValueError("{}, línea {}: {}".format(ruta, n, e)) if o: objetivos.append(o) return objetivos # ───────────────────────── Comprobaciones ───────────────────────── class SinRedirecciones(urllib.request.HTTPRedirectHandler): def redirect_request(self, *a, **k): return None def dias_certificado(host, puerto=443, timeout=8): ctx = ssl.create_default_context() with socket.create_connection((host, puerto), timeout=timeout) as s: with ctx.wrap_socket(s, server_hostname=host) as t: fin = datetime.datetime.strptime(t.getpeercert()["notAfter"], "%b %d %H:%M:%S %Y %Z").replace(tzinfo=datetime.timezone.utc) return (fin - datetime.datetime.now(datetime.timezone.utc)).days def comprobar_http(o, timeout): t0 = time.monotonic() req = urllib.request.Request(o["destino"], headers={"User-Agent": UA}) opener = urllib.request.build_opener(SinRedirecciones) try: r = opener.open(req, timeout=timeout) except urllib.error.HTTPError as e: r = e except ssl.SSLCertVerificationError as e: return {"ok": False, "detalle": "Certificado no válido: {}".format(getattr(e, "verify_message", e)), "ms": None} except (urllib.error.URLError, OSError) as e: motivo = getattr(e, "reason", e) return {"ok": False, "detalle": "Sin respuesta: {}".format(motivo), "ms": None} try: cuerpo = r.read(1024 * 1024) if o.get("contiene") else b"" codigo = getattr(r, "status", None) or r.getcode() finally: r.close() ms = round((time.monotonic() - t0) * 1000) res = {"ok": True, "codigo": codigo, "ms": ms, "detalle": "HTTP {} en {} ms".format(codigo, ms), "avisos": []} if o.get("esperar") is not None and codigo != o["esperar"]: res.update(ok=False, detalle="HTTP {} (se esperaba {})".format(codigo, o["esperar"])) elif o.get("esperar") is None and not 200 <= codigo < 400: res.update(ok=False, detalle="HTTP {}".format(codigo)) elif o.get("contiene") and o["contiene"].encode("utf-8") not in cuerpo: res.update(ok=False, detalle="Responde HTTP {}, pero no contiene «{}»".format(codigo, o["contiene"])) if res["ok"] and o.get("lento") and ms > o["lento"]: res["avisos"].append("Lento: {} ms (límite {} ms)".format(ms, o["lento"])) u = urllib.parse.urlsplit(o["destino"]) if res["ok"] and u.scheme == "https": try: dias = dias_certificado(u.hostname, u.port or 443, timeout) res["cert_dias"] = dias if dias < CERT_AVISO_DIAS: res["avisos"].append("El certificado caduca en {} días".format(dias)) except (OSError, ssl.SSLError, ValueError, KeyError): pass return res def comprobar_tcp(o, timeout): t0 = time.monotonic() try: with socket.create_connection((o["host"], o["puerto"]), timeout=timeout): ms = round((time.monotonic() - t0) * 1000) res = {"ok": True, "ms": ms, "detalle": "Puerto {} abierto en {} ms".format(o["puerto"], ms), "avisos": []} if o.get("lento") and ms > o["lento"]: res["avisos"].append("Lento: {} ms".format(ms)) return res except socket.timeout: return {"ok": False, "ms": None, "detalle": "Puerto {} sin respuesta (filtrado o caído)".format(o["puerto"])} except ConnectionRefusedError: return {"ok": False, "ms": None, "detalle": "Puerto {} cerrado (conexión rechazada)".format(o["puerto"])} except OSError as e: return {"ok": False, "ms": None, "detalle": "No se pudo conectar: {}".format(e)} def comprobar_dns(o, timeout): t0 = time.monotonic() viejo = socket.getdefaulttimeout() socket.setdefaulttimeout(timeout) try: ips = sorted({i[4][0] for i in socket.getaddrinfo(o["host"], None)}) ms = round((time.monotonic() - t0) * 1000) return {"ok": True, "ms": ms, "detalle": "Resuelve a {}".format(", ".join(ips[:3])), "avisos": []} except socket.gaierror as e: return {"ok": False, "ms": None, "detalle": "No se resuelve: {}".format(e.strerror or e)} finally: socket.setdefaulttimeout(viejo) def comprobar(o, timeout=10): f = {"http": comprobar_http, "tcp": comprobar_tcp, "dns": comprobar_dns}[o["tipo"]] r = f(o, timeout) r.setdefault("avisos", []) return r # ───────────────────────── Comprobación global (check-host.net) ───────────────────────── def resumen_global(resultado): """Respuesta de check-result → (ok, total, detalles). Los nodos aún sin respuesta no cuentan.""" ok, total, detalles = 0, 0, [] for nodo, v in (resultado or {}).items(): if v is None: continue total += 1 pais = nodo.split(".")[0][:2].upper() fila = v[0] if isinstance(v, list) and v else None if isinstance(fila, list) and fila and fila[0] == 1: ok += 1 detalles.append("{} OK".format(pais)) else: msg = next((x.get("message") for x in v if isinstance(x, dict)), None) if isinstance(v, list) else None detalles.append("{} {}".format(pais, msg or (fila[2] if isinstance(fila, list) and len(fila) > 2 else "fallo"))) return ok, total, detalles def global_http(url, abrir=urllib.request.urlopen, esperar=8, nodos=5, dormir=time.sleep): """Pregunta a nodos de varios países. Devuelve (ok, total, detalles) o None si el servicio no responde.""" try: q = urllib.parse.urlencode({"host": url, "max_nodes": nodos}) req = urllib.request.Request("https://check-host.net/check-http?" + q, headers={"Accept": "application/json", "User-Agent": UA}) with abrir(req, timeout=15) as r: rid = json.loads(r.read().decode())["request_id"] dormir(esperar) req = urllib.request.Request("https://check-host.net/check-result/" + rid, headers={"Accept": "application/json", "User-Agent": UA}) with abrir(req, timeout=15) as r: return resumen_global(json.loads(r.read().decode())) except (OSError, ValueError, KeyError): return None # ───────────────────────── Avisos ───────────────────────── def mensaje(o, r, antes): if not r["ok"]: txt = "🔴 CAÍDO: {} — {}".format(o["nombre"], r["detalle"]) if r.get("global"): ok, total = r["global"][:2] txt += " · desde otros países: {}/{} lo ven funcionar{}".format(ok, total, " (parece que solo falla desde aquí)" if ok and ok == total else "") return txt if antes and not antes.get("ok"): desde = antes.get("desde") dur = "" if desde: segs = int(time.time() - desde) dur = " tras {} min caído".format(max(1, segs // 60)) return "🟢 RECUPERADO: {}{} — {}".format(o["nombre"], dur, r["detalle"]) return None def cuerpo_webhook(url, texto): host = urllib.parse.urlsplit(url).hostname or "" if "discord" in host: return {"content": texto} return {"text": texto} # Slack, Microsoft Teams (webhook entrante) y la mayoría de sistemas def enviar_webhook(url, texto, abrir=urllib.request.urlopen): datos = json.dumps(cuerpo_webhook(url, texto)).encode("utf-8") req = urllib.request.Request(url, data=datos, headers={"Content-Type": "application/json", "User-Agent": UA}, method="POST") try: with abrir(req, timeout=10) as r: return 200 <= r.status < 300 except (OSError, ValueError): return False # ───────────────────────── Ronda ───────────────────────── def ronda(objetivos, estado, args, comprobar_fn=comprobar, global_fn=global_http, webhook_fn=enviar_webhook, salida=print): """Comprueba todo una vez. Devuelve (resultados, mensajes de cambio).""" resultados, cambios = [], [] ahora = time.time() for o in objetivos: r = comprobar_fn(o, args.timeout) if not r["ok"] and args.globl and o["tipo"] == "http": g = global_fn(o["destino"]) if g: r["global"] = g antes = estado.get(o["destino"]) cambio = antes is None and not r["ok"] or (antes is not None and antes.get("ok") != r["ok"]) txt = mensaje(o, r, antes) if cambio else None if txt: cambios.append(txt) if args.webhook: webhook_fn(args.webhook, txt) nuevo = {"ok": r["ok"], "detalle": r["detalle"], "comprobado": ahora} nuevo["desde"] = (antes.get("desde") if antes and not antes.get("ok") and not r["ok"] else (ahora if not r["ok"] else None)) estado[o["destino"]] = nuevo resultados.append((o, r)) marca = "✖" if not r["ok"] else "▲" if r["avisos"] else "✓" salida(" {} {:<32} {}{}".format(marca, o["nombre"][:32], r["detalle"], (" · " + " · ".join(r["avisos"])) if r["avisos"] else "")) return resultados, cambios def codigo_salida(resultados): if any(not r["ok"] for _, r in resultados): return 2 if any(r["avisos"] for _, r in resultados): return 1 return 0 def guardar_csv(ruta, resultados): nuevo = not os.path.exists(ruta) or os.path.getsize(ruta) == 0 with open(ruta, "a", newline="", encoding="utf-8") as f: w = csv.writer(f) if nuevo: w.writerow(["fecha", "objetivo", "estado", "ms", "detalle", "avisos"]) fecha = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") for o, r in resultados: w.writerow([fecha, o["nombre"], "ok" if r["ok"] else "caido", r.get("ms") or "", r["detalle"], " · ".join(r["avisos"])]) def preparar_salida(): for flujo in (sys.stdout, sys.stderr): try: if not flujo.isatty(): flujo.reconfigure(encoding="utf-8", errors="replace") else: flujo.reconfigure(errors="replace") except (AttributeError, ValueError): pass def main(argv=None, **inyectado): preparar_salida() ap = argparse.ArgumentParser(description="Vigila webs, puertos y nombres DNS y avisa cuando cambian de estado.") ap.add_argument("objetivos", nargs="*", help="https://…, tcp:host:puerto o dns:nombre") ap.add_argument("--lista", metavar="FICHERO", help="fichero con un objetivo por línea (# para comentarios)") ap.add_argument("--cada", type=int, metavar="SEG", help="repetir cada SEG segundos (sin esto, una sola ronda: ideal para cron)") ap.add_argument("--estado", metavar="FICHERO", default=".vigilante-estado.json", help="dónde recordar el estado entre ejecuciones") ap.add_argument("--webhook", metavar="URL", help="webhook de Slack, Discord o Teams para los avisos") ap.add_argument("--global", dest="globl", action="store_true", help="si algo falla, comprobarlo desde varios países (check-host.net)") ap.add_argument("--csv", metavar="FICHERO", help="añadir cada ronda a un historial CSV") ap.add_argument("--timeout", type=float, default=10, help="segundos de espera por comprobación (10)") a = ap.parse_args(argv) try: objetivos = [o for o in (parse_objetivo(x) for x in a.objetivos) if o] if a.lista: objetivos += leer_lista(a.lista) except (ValueError, OSError) as e: print("Error: {}".format(e), file=sys.stderr) return 3 if not objetivos: print("Indica al menos un objetivo o --lista. Ejemplo: python vigilar_servicios.py https://miweb.com", file=sys.stderr) return 3 estado = {} if os.path.exists(a.estado): try: with open(a.estado, encoding="utf-8") as f: estado = json.load(f) except (OSError, ValueError): estado = {} codigo = 0 while True: print("{} · {} objetivo{}".format(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), len(objetivos), "" if len(objetivos) == 1 else "s")) resultados, cambios = ronda(objetivos, estado, a, **inyectado) for c in cambios: print(c) with open(a.estado, "w", encoding="utf-8") as f: json.dump(estado, f, ensure_ascii=False, indent=2) if a.csv: guardar_csv(a.csv, resultados) codigo = codigo_salida(resultados) if not a.cada: return codigo try: time.sleep(max(5, a.cada)) except KeyboardInterrupt: return codigo if __name__ == "__main__": sys.exit(main())