chemavxandClaude Opus 4.8
07c872e847
Deja en el repo lo que hoy destapó que argocd-redis, el daemon.json de gitea y siete imágenes llevaban meses diciendo una cosa mientras el cluster hacía otra. Compara cada manifiesto con el objeto vivo, en las dos direcciones (el fichero afirma algo falso / hay algo desplegado que el fichero no menciona / el objeto no existe). Sólo mira los directorios fuera de ArgoCD, que es donde no hay nadie comprobando: en los demás ArgoCD ya marca OutOfSync cada 3 minutos. La lista de directorios se deduce preguntando a ArgoCD por los paths de sus Applications, así que si mañana un namespace entra en GitOps el script se ajusta solo. Las excepciones conocidas van en excepciones-auditoria.yaml CON su motivo, para que lo normal sea "hallazgos NUEVOS: 0" y cualquier cosa que asome merezca una mirada. Un informe que siempre saca ruido deja de leerse — que es justo el fallo que se ha estado corrigiendo estos días. El script también avisa cuando una excepción ya no se dispara: se arregló y la entrada sobra. Probado viéndolo fallar, no sólo pasar: inyectadas a propósito una avería de cada tipo (imagen cambiada, env borrada del fichero, objeto inexistente) y las caza las tres saliendo con código 1. Al probarlo apareció un fallo propio: con -d marcaba como sobrantes las excepciones de los directorios no barridos; ahora sólo lo afirma tras un barrido completo. No toca nada (sólo lectura), nunca compara valores de Secret —sólo nombres de clave— y se salta lo que ignora .gitignore. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
255 lines
11 KiB
Python
Executable File
255 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Compara los manifiestos de este repo con lo que hay desplegado de verdad.
|
|
|
|
Sólo mira los directorios que NO gestiona ArgoCD, que es donde nadie comprueba
|
|
nada: allí git es documentación, y una documentación desfasada es peor que no
|
|
tenerla — se aplica creyendo que restaura y en realidad rompe. Los directorios
|
|
bajo ArgoCD se saltan porque el propio ArgoCD ya los compara cada 3 minutos.
|
|
|
|
Mira en las dos direcciones, porque fallan distinto:
|
|
|
|
DIFIERE el fichero afirma algo que ya no es cierto
|
|
SOLO VIVO hay algo desplegado que el fichero no menciona
|
|
NO DESPLEGADO el fichero describe un objeto que no existe
|
|
|
|
Las excepciones conocidas van en excepciones-auditoria.yaml, con su motivo, para
|
|
que la salida normal sea "0 hallazgos nuevos" y cualquier cosa que aparezca
|
|
merezca una mirada. Si una excepción deja de darse, también lo dice: sobra.
|
|
|
|
Uso:
|
|
python3 tools/audita-exports.py # informe
|
|
python3 tools/audita-exports.py -v # incluye las excepciones
|
|
python3 tools/audita-exports.py -d gitea # sólo esos directorios
|
|
|
|
Sale con 1 si hay hallazgos nuevos, para poder colgarlo de un temporizador.
|
|
Nunca imprime valores de Secret: de esos compara sólo los nombres de clave.
|
|
"""
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
RAIZ = Path(__file__).resolve().parent.parent
|
|
|
|
# Ruido: lo pone el API server o un controlador, no es configuración deseada.
|
|
META_FUERA = {'creationTimestamp', 'resourceVersion', 'uid', 'generation',
|
|
'managedFields', 'selfLink', 'ownerReferences', 'finalizers'}
|
|
ANOT_FUERA = {'kubectl.kubernetes.io/last-applied-configuration',
|
|
'deployment.kubernetes.io/revision',
|
|
'kubectl.kubernetes.io/restartedAt',
|
|
'deprecated.daemonset.template.generation',
|
|
'endpoints.kubernetes.io/last-change-trigger-time',
|
|
'control-plane.alpha.kubernetes.io/leader'}
|
|
# Campos que el API server rellena solo y que nadie escribe a mano.
|
|
DEFECTO = {'clusterIP', 'clusterIPs', 'ipFamilies', 'ipFamilyPolicy',
|
|
'internalTrafficPolicy', 'sessionAffinity', 'nodePort',
|
|
'terminationMessagePath', 'terminationMessagePolicy', 'dnsPolicy',
|
|
'schedulerName', 'restartPolicy', 'defaultMode', 'revisionHistoryLimit',
|
|
'progressDeadlineSeconds', 'protocol', 'targetPort'}
|
|
|
|
|
|
def limpia(o, en_meta=False):
|
|
if isinstance(o, dict):
|
|
r = {}
|
|
for k, v in o.items():
|
|
if (k == 'status' and not en_meta) or k in META_FUERA or k in DEFECTO:
|
|
continue
|
|
if k == 'annotations' and isinstance(v, dict):
|
|
v = {a: b for a, b in v.items() if a not in ANOT_FUERA}
|
|
if not v:
|
|
continue
|
|
r[k] = limpia(v, en_meta or k == 'metadata')
|
|
return r
|
|
if isinstance(o, list):
|
|
return [limpia(x, en_meta) for x in o]
|
|
return o
|
|
|
|
|
|
def compara(fich, vivo, ruta=''):
|
|
"""Todo lo que el fichero afirma, ¿sigue siendo verdad en el cluster?"""
|
|
out = []
|
|
if isinstance(fich, dict):
|
|
if not isinstance(vivo, dict):
|
|
return [(ruta, fich, vivo)]
|
|
for k, v in fich.items():
|
|
if k not in vivo:
|
|
out.append((f'{ruta}.{k}', v, '<ausente en el cluster>'))
|
|
else:
|
|
out += compara(v, vivo[k], f'{ruta}.{k}')
|
|
elif isinstance(fich, list):
|
|
if not isinstance(vivo, list):
|
|
return [(ruta, fich, vivo)]
|
|
# listas de dicts con nombre (containers, env, volumes...): emparejar por
|
|
# nombre, no por posición, o reordenar el fichero saldría como deriva
|
|
if fich and all(isinstance(x, dict) and 'name' in x for x in fich):
|
|
porn = {x['name']: x for x in vivo if isinstance(x, dict) and 'name' in x}
|
|
for x in fich:
|
|
if x['name'] not in porn:
|
|
out.append((f"{ruta}[{x['name']}]", 'presente', '<ausente en el cluster>'))
|
|
else:
|
|
out += compara(x, porn[x['name']], f"{ruta}[{x['name']}]")
|
|
elif fich != vivo:
|
|
out.append((ruta, fich, vivo))
|
|
elif fich != vivo:
|
|
out.append((ruta, fich, vivo))
|
|
return out
|
|
|
|
|
|
def perfil(o):
|
|
"""Lo esencial de una carga de trabajo, para ver qué hay vivo y sin escribir."""
|
|
spec = (o.get('spec') or {}).get('template', {}).get('spec')
|
|
if not spec:
|
|
return None
|
|
p = {'volumenes': sorted(v.get('name', '?') for v in spec.get('volumes') or [])}
|
|
for c in (spec.get('containers') or []) + (spec.get('initContainers') or []):
|
|
p[f"contenedor:{c['name']}"] = {
|
|
'env': sorted(e.get('name', '?') for e in c.get('env') or []),
|
|
'montajes': sorted(m.get('name', '?') for m in c.get('volumeMounts') or []),
|
|
}
|
|
return p
|
|
|
|
|
|
def dirs_de_argocd():
|
|
r = subprocess.run(['kubectl', 'get', 'app', '-n', 'argocd', '-o',
|
|
'jsonpath={range .items[*]}{.spec.source.path}{"\\n"}{end}'],
|
|
capture_output=True, text=True)
|
|
return {p.strip().strip('/') for p in r.stdout.split('\n') if p.strip() not in ('', '.')}
|
|
|
|
|
|
def ignorados_por_git():
|
|
r = subprocess.run(['git', 'ls-files', '--others', '--ignored', '--exclude-standard'],
|
|
cwd=RAIZ, capture_output=True, text=True)
|
|
return {l.strip() for l in r.stdout.split('\n') if l.strip()}
|
|
|
|
|
|
def vivo_de(kind, nombre, ns):
|
|
orden = ['kubectl', 'get', kind, nombre, '-o', 'json'] + (['-n', ns] if ns else [])
|
|
r = subprocess.run(orden, capture_output=True, text=True)
|
|
return json.loads(r.stdout) if r.returncode == 0 else None
|
|
|
|
|
|
def carga_excepciones():
|
|
f = RAIZ / 'tools' / 'excepciones-auditoria.yaml'
|
|
if not f.exists():
|
|
return []
|
|
return (yaml.safe_load(open(f)) or {}).get('excepciones') or []
|
|
|
|
|
|
def es_esperado(exc, fichero, tipo, que):
|
|
for e in exc:
|
|
if not fnmatch.fnmatch(fichero, e['fichero']):
|
|
continue
|
|
if e.get('tipo') and e['tipo'] != tipo:
|
|
continue
|
|
if e.get('contiene') and e['contiene'] not in que:
|
|
continue
|
|
return e
|
|
return None
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument('-d', '--dirs', nargs='*', help='limitar a estos directorios')
|
|
ap.add_argument('-v', '--verbose', action='store_true', help='mostrar también las excepciones')
|
|
a = ap.parse_args()
|
|
|
|
bajo_argocd = dirs_de_argocd()
|
|
if a.dirs:
|
|
objetivo = [d for d in a.dirs]
|
|
else:
|
|
objetivo = sorted(p.name for p in RAIZ.iterdir()
|
|
if p.is_dir() and not p.name.startswith('.')
|
|
and p.name not in bajo_argocd and p.name != 'tools')
|
|
ignorados = ignorados_por_git()
|
|
exc = carga_excepciones()
|
|
|
|
nuevos, esperados, revisados = [], [], 0
|
|
for d in objetivo:
|
|
for f in sorted((RAIZ / d).glob('*.yaml')):
|
|
rel = str(f.relative_to(RAIZ))
|
|
if rel in ignorados:
|
|
continue
|
|
try:
|
|
docs = [x for x in yaml.safe_load_all(open(f)) if x]
|
|
except Exception as e:
|
|
nuevos.append((rel, 'ILEGIBLE', str(e), ''))
|
|
continue
|
|
planos = []
|
|
for doc in docs:
|
|
planos += doc.get('items', []) if doc.get('kind') == 'List' else [doc]
|
|
for doc in planos:
|
|
kind, meta = doc.get('kind'), doc.get('metadata') or {}
|
|
nombre = meta.get('name')
|
|
if not kind or not nombre:
|
|
continue
|
|
ns = meta.get('namespace') or (None if kind in ('Namespace',) else d)
|
|
revisados += 1
|
|
vivo = vivo_de(kind, nombre, ns)
|
|
hall = []
|
|
if vivo is None:
|
|
hall.append(('NO DESPLEGADO', f'{kind}/{nombre}', ''))
|
|
else:
|
|
if kind == 'Secret': # jamás comparar valores
|
|
for o in (doc, vivo):
|
|
for campo in ('data', 'stringData'):
|
|
if campo in o:
|
|
o[campo] = {k: '<redactado>' for k in o[campo]}
|
|
A, B = limpia(doc), limpia(vivo)
|
|
for ruta, vf, vv in compara(A, B):
|
|
hall.append(('DIFIERE', f'{kind}/{nombre}{ruta}', f'git={vf!r} vivo={vv!r}'))
|
|
pf, pv = perfil(A), perfil(B)
|
|
if pf and pv:
|
|
for clave, val in pv.items():
|
|
if clave not in pf:
|
|
hall.append(('SOLO VIVO', f'{kind}/{nombre} {clave}', ''))
|
|
elif isinstance(val, dict):
|
|
for sub in ('env', 'montajes'):
|
|
extra = set(val[sub]) - set(pf[clave].get(sub) or [])
|
|
if extra:
|
|
hall.append(('SOLO VIVO', f'{kind}/{nombre} {clave}.{sub}',
|
|
', '.join(sorted(extra))))
|
|
else:
|
|
extra = set(val) - set(pf.get(clave) or [])
|
|
if extra:
|
|
hall.append(('SOLO VIVO', f'{kind}/{nombre} {clave}',
|
|
', '.join(sorted(extra))))
|
|
for tipo, que, det in hall:
|
|
e = es_esperado(exc, rel, tipo, que)
|
|
(esperados if e else nuevos).append((rel, tipo, que, det, e))
|
|
|
|
print(f'objetos revisados: {revisados} directorios: {len(objetivo)} '
|
|
f'(fuera de ArgoCD) excepciones conocidas: {len(exc)}\n')
|
|
for h in nuevos:
|
|
rel, tipo, que, det = h[0], h[1], h[2], h[3]
|
|
print(f'{tipo:<14} {rel}\n{"":<14} {que}')
|
|
if det:
|
|
print(f'{"":<14} {det[:220]}')
|
|
if a.verbose and esperados:
|
|
print('\n--- esperados (justificados en tools/excepciones-auditoria.yaml) ---')
|
|
for rel, tipo, que, det, e in esperados:
|
|
print(f'{tipo:<14} {rel} [{e["motivo"]}]')
|
|
|
|
# Una excepción que ya no se dispara es una excepción que sobra — pero eso
|
|
# sólo se puede afirmar tras un barrido completo: con -d quedan fuera
|
|
# directorios enteros y todas sus excepciones parecerían muertas.
|
|
usadas = {id(e) for *_, e in esperados}
|
|
muertas = [] if a.dirs else [e for e in exc if id(e) not in usadas]
|
|
|
|
print(f'\nhallazgos NUEVOS: {len(nuevos)} esperados: {len(esperados)}')
|
|
if muertas:
|
|
print(f'excepciones que ya no hacen falta ({len(muertas)}), bórralas:')
|
|
for e in muertas:
|
|
print(f' - {e["fichero"]} [{e["motivo"]}]')
|
|
if not nuevos:
|
|
print('sin deriva: los manifiestos describen lo desplegado')
|
|
return 1 if nuevos else 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|