k8s-manifests/tools/audita-semanal.py

#!/usr/bin/env python3
"""Pasa la auditoría de exports una vez por semana y avisa por Telegram.

Callado por diseño: sólo escribe si hay hallazgos NUEVOS y además cambian
respecto a la semana anterior. Si la deriva sigue igual porque aún no se ha
arreglado, no repite el mensaje — un aviso que llega todas las semanas deja de
leerse, y entonces no sirve para nada.

Avisa también de dos cosas que invalidarían la propia auditoría:
  - el repo local por detrás de origin  -> compararía ficheros viejos
  - excepciones que ya no se disparan   -> justificaciones muertas acumulándose

Correlato de tools/audita-exports.py, que es quien hace el trabajo. Aquí sólo
está la decisión de molestar o no.
"""
import hashlib
import json
import os
import subprocess
import sys
import urllib.parse
import urllib.request
from pathlib import Path

RAIZ = Path(__file__).resolve().parent.parent
ESTADO = Path(os.path.expanduser('~/.local/state/audita-exports/last.json'))
# Mismo secret que usan seo-watch y hermes-bot. Al vivir en Infisical, una
# rotación llega sola a los tres; no hay copia del token en disco.
TG_SECRET = ('monitoring', 'grafana-telegram-infisical')


def sh(cmd, cwd=None):
    return subprocess.run(cmd, shell=True, capture_output=True, text=True, cwd=cwd)


def telegram(texto):
    def leer(clave):
        r = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} "
               f"-o jsonpath='{{.data.{clave}}}'")
        if r.returncode != 0 or not r.stdout:
            raise RuntimeError(f'no pude leer {clave} del secret: {r.stderr.strip()[:120]}')
        import base64
        return base64.b64decode(r.stdout).decode()

    datos = urllib.parse.urlencode({'chat_id': leer('TELEGRAM_CHAT_ID'),
                                    'text': texto[:3900]}).encode()
    pet = urllib.request.Request(
        f"https://api.telegram.org/bot{leer('TELEGRAM_BOT_TOKEN')}/sendMessage", data=datos)
    with urllib.request.urlopen(pet, timeout=30) as resp:
        return json.loads(resp.read()).get('ok', False)


def estado_del_repo():
    """Devuelve (motivos_para_avisar, contexto).

    Ir por detrás de origin SÍ invalida la auditoría: se estarían comparando
    ficheros viejos contra el cluster. Tener cambios sin commitear NO — es
    estado de trabajo normal, la copia local es la intención actual y es
    exactamente lo que hay que comparar. Avisar de eso convertiría el vigilante
    en ruido semanal, que es como muere un vigilante.
    """
    motivos, contexto = [], []
    sh('git fetch --quiet', cwd=RAIZ)
    detras = sh('git rev-list --count HEAD..@{u}', cwd=RAIZ).stdout.strip()
    if detras.isdigit() and int(detras) > 0:
        motivos.append(f'el repo local va {detras} commit(s) por detrás de origin: '
                       'la comparación se ha hecho con ficheros viejos')
    if sh('git status --porcelain', cwd=RAIZ).stdout.strip():
        contexto.append('(el repo tiene cambios sin commitear)')
    return motivos, contexto


def main():
    r = sh(f'{sys.executable} {RAIZ}/tools/audita-exports.py', cwd=RAIZ)
    salida = (r.stdout or '') + (r.stderr or '')

    # El auditor se rompió: eso también es noticia, y no debe pasar por "sin deriva".
    # No basta el código de salida: un error de sintaxis en Python también sale con
    # 1, que es justo el código de "hay deriva". Se exige además que el informe
    # tenga su línea de resumen; si no está, es que no llegó a terminar.
    if r.returncode not in (0, 1) or 'hallazgos NUEVOS:' not in salida:
        telegram(f'🔴 Auditoría de manifiestos: el propio auditor ha fallado '
                 f'(código {r.returncode}).\n\n{salida[-1500:]}')
        return 2

    hay_deriva = r.returncode == 1
    sobrantes = 'excepciones que ya no hacen falta' in salida
    motivos_repo, contexto_repo = estado_del_repo()

    if not (hay_deriva or sobrantes or motivos_repo):
        print('sin novedad: nada que avisar')
        ESTADO.parent.mkdir(parents=True, exist_ok=True)
        ESTADO.write_text(json.dumps({'huella': None}))
        return 0

    cuerpo = salida.strip()
    huella = hashlib.sha256((cuerpo + '|'.join(motivos_repo)).encode()).hexdigest()
    try:
        previo = json.loads(ESTADO.read_text()).get('huella')
    except (FileNotFoundError, ValueError):
        previo = None

    if huella == previo:
        print('mismos hallazgos que la semana pasada: no repito el aviso')
        return 0

    cabecera = '🟠 Auditoría de manifiestos: los ficheros y el cluster no coinciden.'
    if not hay_deriva:
        cabecera = '🟡 Auditoría de manifiestos: sin deriva, pero hay algo que mirar.'
    extra = ''.join(f'\n⚠️ {m}' for m in motivos_repo) + ''.join(f'\n{c}' for c in contexto_repo)
    recordatorio = ('\n\nRecuerda: en estos directorios un commit NO despliega. '
                    'Hay que editar el fichero Y hacer kubectl apply.')
    telegram(f'{cabecera}{extra}\n\n{cuerpo}{recordatorio}')

    ESTADO.parent.mkdir(parents=True, exist_ok=True)
    ESTADO.write_text(json.dumps({'huella': huella}))
    print('aviso enviado')
    return 0


if __name__ == '__main__':
    sys.exit(main())