#!/usr/bin/env python3 """ Auditor de JWT — Ciberkit =============================== Decodifica y audita tokens JWT, comprueba si un token HS256/384/512 usa un secreto típico y busca tokens filtrados en logs, volcados HAR, ficheros .env o carpetas enteras. Úsalo SOLO con tokens y sistemas propios o con autorización expresa. Sin dependencias: Python 3.8+ en Windows, Linux o macOS. Ejemplos -------- python jwt_auditar.py eyJhbGciOi... # audita un token echo "Bearer eyJ..." | python jwt_auditar.py # desde la entrada estándar python jwt_auditar.py eyJ... --secreto "mi-secreto" # verifica la firma HS python jwt_auditar.py eyJ... --secretos lista.txt # añade secretos a probar python jwt_auditar.py --buscar /var/log/nginx app.har # tokens filtrados en logs python jwt_auditar.py --buscar logs/ --json informe.json Código de salida: 0 sin problemas, 1 avisos, 2 críticos (útil en CI). """ import argparse import base64 import hashlib import hmac import json import os import re import sys import time FORMATO = "suite-redes.jwt-auditoria/1" MAX_FICHERO = 50 * 1024 * 1024 RE_JWT = re.compile(r"eyJ[A-Za-z0-9_-]{5,}\.eyJ[A-Za-z0-9_-]{5,}\.[A-Za-z0-9_-]*") RE_SENSIBLE = re.compile(r"pass(word|wd)?|pwd|secret|api[_-]?key|private|credit|card|tarjeta|cvv|ssn|dni|nif|iban|clave", re.I) RE_ASIMETRICO = re.compile(r"^(RS|PS|ES)(256|384|512)$|^EdDSA$|^Ed25519$") HASHES = {"HS256": hashlib.sha256, "HS384": hashlib.sha384, "HS512": hashlib.sha512} ORDEN = {"critico": 0, "aviso": 1, "info": 2, "ok": 3} ETIQUETA = {"critico": "CRÍTICO", "aviso": "AVISO", "info": "INFO", "ok": "OK"} # Misma lista que la web (assets/js/core/jwt.js → WEAK_SECRETS); los tests comprueban que coinciden. SECRETOS_DEBILES = [ "secret", "your-256-bit-secret", "your-384-bit-secret", "your-512-bit-secret", "secretkey", "secret_key", "secret-key", "mysecret", "my_secret", "my-secret", "mysecretkey", "my_secret_key", "jwt", "jwtsecret", "jwt_secret", "jwt-secret", "jwtSecret", "JWT_SECRET", "jwt_secret_key", "jwtkey", "jwt-key", "key", "private", "privatekey", "password", "Password", "password123", "pass", "123456", "12345678", "1234567890", "admin", "changeme", "change_me", "changeit", "default", "test", "testing", "test123", "dev", "development", "qwerty", "letmein", "welcome", "hello", "example", "demo", "token", "tokensecret", "auth", "authsecret", "auth_secret", "app_secret", "appsecret", "application_secret", "api_secret", "apisecret", "apikey", "api_key", "supersecret", "super_secret", "super-secret", "supersecretkey", "topsecret", "shhhhh", "shhhhhared-secret", "keyboard cat", "keyboardcat", "node", "express", "nodejs", "django", "flask", "spring", "laravel", "rails", "secret123", "Secret123", "S3cr3t", "s3cr3t", "secretpassword", "mypassword", "pa$$word", "hmac", "hs256", "HS256", "signature", "sign", "signing_key", "signingkey", "encryption_key", "session", "session_secret", "cookie", "gogogo", "abc123", "abcdef", "ninja", "master", "root", "toor", "user", "guest", "null", "undefined", "none", "", "secreto", "clave", "clavesecreta", "clave_secreta", "contraseña", "contrasena", "miclave", "misecreto", ] class ErrorJWT(ValueError): pass # ───────────────────────── Decodificar ───────────────────────── def b64url_decode(parte): if not re.fullmatch(r"[A-Za-z0-9_-]*", parte): raise ErrorJWT("contiene caracteres que no son Base64URL") if len(parte) % 4 == 1: raise ErrorJWT("longitud no válida para Base64URL") return base64.urlsafe_b64decode(parte + "=" * (-len(parte) % 4)) def limpiar(token): token = token.strip() token = re.sub(r"^(authorization:\s*)?bearer\s+", "", token, flags=re.I) return re.sub(r"\s+", "", token.strip("\"'")) def _json(parte, nombre): try: valor = json.loads(b64url_decode(parte).decode("utf-8")) except ErrorJWT as e: raise ErrorJWT(f"{nombre}: {e}") from None except (UnicodeDecodeError, ValueError): raise ErrorJWT(f"{nombre}: no es un objeto JSON") from None if not isinstance(valor, dict): raise ErrorJWT(f"{nombre}: no es un objeto JSON") return valor def decodificar(token): token = limpiar(token) partes = token.split(".") if len(partes) == 5: return {"tipo": "JWE", "cabecera": _json(partes[0], "Cabecera"), "contenido": None, "firma": b"", "entrada": "", "token": token} if len(partes) != 3: raise ErrorJWT(f"un JWT tiene 3 partes separadas por puntos y este tiene {len(partes)}") return { "tipo": "JWS", "cabecera": _json(partes[0], "Cabecera"), "contenido": _json(partes[1], "Contenido"), "firma": b64url_decode(partes[2]), "entrada": f"{partes[0]}.{partes[1]}", "token": token, } # ───────────────────────── Auditoría ───────────────────────── def duracion(segundos): s = abs(segundos) for tam, uno, varios in ((86400 * 365, "año", "años"), (86400 * 30, "mes", "meses"), (86400, "día", "días"), (3600, "hora", "horas"), (60, "minuto", "minutos")): if s >= tam: n = int(s // tam) return f"{n} {uno if n == 1 else varios}" return "unos segundos" def auditar(d, ahora=None): ahora = time.time() if ahora is None else ahora out = [] def add(sev, texto, ref=None): out.append({"severidad": sev, "texto": texto, "ref": ref}) h, p = d["cabecera"], d["contenido"] alg = h.get("alg") if d["tipo"] == "JWE": add("ok", f"Token cifrado (JWE, {alg} + {h.get('enc')}).", "RFC 7516") return out if alg is None: add("critico", "La cabecera no indica el algoritmo (alg).", "RFC 7515 §4.1.1") elif str(alg).lower() == "none": add("critico", "alg «none»: el token no tiene firma y cualquiera puede fabricarlo.", "RFC 8725 §3.1") elif alg in HASHES: add("info", f"{alg} usa un secreto compartido: quien verifica también puede firmar.", "RFC 8725 §3.5") elif RE_ASIMETRICO.match(str(alg)): add("ok", f"Firma asimétrica ({alg}).", "RFC 8725 §3.1") else: add("aviso", f"Algoritmo poco común: {alg}.", "RFC 7518") if not d["firma"] and alg and str(alg).lower() != "none": add("critico", "La firma está vacía aunque la cabecera dice que está firmado.", "RFC 7515") if "jwk" in h: add("critico", "La cabecera trae su propia clave pública (jwk): si el servidor la usa, cualquiera puede firmar.", "CVE-2018-0114 · RFC 8725 §3.10") for k in ("jku", "x5u"): if h.get(k): add("aviso", f"La cabecera indica de dónde descargar la clave ({k}: {h[k]}). Solo deben aceptarse URLs de una lista permitida.", "RFC 8725 §3.10") kid = h.get("kid") if isinstance(kid, str) and re.search(r"\.\.|[/\\'\";|`$]|\s", kid): add("aviso", f"El «kid» contiene caracteres sospechosos ({kid}): posible inyección o recorrido de rutas.", "RFC 8725 §3.10") def num(v): return isinstance(v, (int, float)) and not isinstance(v, bool) exp, iat, nbf = p.get("exp"), p.get("iat"), p.get("nbf") if exp is None: add("aviso", "Sin caducidad (exp): si alguien lo roba, le sirve para siempre.", "RFC 8725 §3.10") elif not num(exp): add("aviso", "exp no es un número de segundos (NumericDate).", "RFC 7519 §4.1.4") elif exp < ahora: add("info", f"Caducado hace {duracion(ahora - exp)}.") else: add("ok", f"Válido durante {duracion(exp - ahora)} más.") if num(exp) and num(iat): vida = exp - iat if vida <= 0: add("aviso", "exp es anterior o igual a iat: el token nació caducado.") elif vida > 86400: add("aviso", f"Vida muy larga: {duracion(vida)}. Para tokens de acceso se recomiendan minutos (5–60).", "RFC 9700 §2.2") elif vida > 3600: add("info", f"Vida de {duracion(vida)}.") if num(iat) and iat > ahora + 300: add("aviso", f"Emitido en el futuro (dentro de {duracion(iat - ahora)}): ¿reloj desajustado?") if num(nbf) and nbf > ahora: add("info", f"Aún no es válido (empieza dentro de {duracion(nbf - ahora)}).") if "aud" not in p: add("aviso", "Sin audiencia (aud): podría reutilizarse en otra API que confíe en el mismo emisor.", "RFC 8725 §3.9") if "iss" not in p: add("info", "Sin emisor (iss).", "RFC 8725 §3.8") fugas = [] def recorrer(obj, ruta): for k, v in obj.items(): if RE_SENSIBLE.search(k) and v not in (None, "") and not isinstance(v, bool): fugas.append(ruta + k) if isinstance(v, dict): recorrer(v, f"{ruta}{k}.") recorrer(p, "") if fugas: add("critico", f"Posibles datos sensibles en el contenido ({', '.join(fugas)}). Un JWT firmado no está cifrado.", "RFC 8725 §3.6") return sorted(out, key=lambda f: ORDEN[f["severidad"]]) # ───────────────────────── Firma HS ───────────────────────── def firma_hs(alg, secreto, entrada): return hmac.new(secreto, entrada.encode("ascii"), HASHES[alg]).digest() def verificar_hs(d, secreto): alg = d["cabecera"].get("alg") if alg not in HASHES: raise ErrorJWT(f"solo se verifican tokens HS256/384/512 (este es {alg}); para RS/ES usa la web o tu librería") if isinstance(secreto, str): secreto = secreto.encode("utf-8") return hmac.compare_digest(firma_hs(alg, secreto, d["entrada"]), d["firma"]) def buscar_secreto(d, extra=()): """Devuelve el secreto si es uno de los típicos (o de la lista extra); None si no.""" alg = d["cabecera"].get("alg") if alg not in HASHES or d["tipo"] != "JWS": return None vistos = set() for s in list(SECRETOS_DEBILES) + list(extra): if s in vistos: continue vistos.add(s) if hmac.compare_digest(firma_hs(alg, s.encode("utf-8"), d["entrada"]), d["firma"]): return s return None # ───────────────────────── Buscar en ficheros ───────────────────────── def huella(token): return hashlib.sha256(token.encode("ascii")).hexdigest()[:12] def enmascarar(token): return f"{token[:16]}…{token[-6:]}" if len(token) > 30 else token def ficheros(rutas): for ruta in rutas: if os.path.isdir(ruta): for raiz, dirs, archivos in os.walk(ruta): dirs[:] = [x for x in dirs if x not in (".git", "node_modules", "__pycache__")] for a in sorted(archivos): yield os.path.join(raiz, a) else: yield ruta def buscar_en_ficheros(rutas): """Devuelve {token: [(fichero, línea), …]} sin repetir tokens.""" hallados = {} for f in ficheros(rutas): try: if os.path.getsize(f) > MAX_FICHERO: continue with open(f, "r", encoding="utf-8", errors="replace") as fh: for n, linea in enumerate(fh, 1): if "eyJ" not in linea: continue for t in RE_JWT.findall(linea): hallados.setdefault(t, []).append((f, n)) except OSError: continue return hallados # ───────────────────────── Informe ───────────────────────── def analizar(token, extra=(), secreto=None, ahora=None): d = decodificar(token) hallazgos = auditar(d, ahora) r = {"cabecera": d["cabecera"], "contenido": d["contenido"], "hallazgos": hallazgos, "huella": huella(d["token"])} if d["tipo"] == "JWS" and d["cabecera"].get("alg") in HASHES: s = buscar_secreto(d, extra) r["secreto_debil"] = s is not None if s is not None: hallazgos.insert(0, {"severidad": "critico", "texto": f"Firmado con un secreto típico: «{s}». Cualquiera puede fabricar tokens válidos.", "ref": "RFC 8725 §3.5"}) if secreto is not None: r["firma_valida"] = verificar_hs(d, secreto) if not r["firma_valida"]: hallazgos.insert(0, {"severidad": "aviso", "texto": "La firma NO coincide con el secreto indicado.", "ref": None}) else: hallazgos.append({"severidad": "ok", "texto": "Firma válida con el secreto indicado.", "ref": None}) return r def peor(hallazgos): return min((ORDEN[f["severidad"]] for f in hallazgos), default=3) def imprimir(r, titulo): print(f"\n{titulo}") c = r["contenido"] or {} resumen = ", ".join(f"{k}={c[k]}" for k in ("iss", "sub", "aud") if k in c) print(f" alg={r['cabecera'].get('alg')} {resumen}") for f in r["hallazgos"]: ref = f" [{f['ref']}]" if f.get("ref") else "" print(f" {ETIQUETA[f['severidad']]:<8} {f['texto']}{ref}") def preparar_salida(): """Evita que el script falle en consolas o redirecciones sin UTF-8 (p. ej. cp1252 en Windows).""" 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): preparar_salida() ap = argparse.ArgumentParser(description="Audita tokens JWT y busca tokens filtrados en ficheros. Solo para sistemas propios o autorizados.") ap.add_argument("token", nargs="?", help="JWT (o «Bearer …»). Si no se indica, se lee de la entrada estándar") ap.add_argument("--buscar", nargs="+", metavar="RUTA", help="ficheros o carpetas donde buscar JWT (logs, HAR, .env…)") ap.add_argument("--secreto", help="secreto HS para verificar la firma") ap.add_argument("--secretos", metavar="FICHERO", help="lista de secretos adicionales a probar (uno por línea)") ap.add_argument("--json", metavar="FICHERO", help="guarda el informe en JSON") args = ap.parse_args(argv) extra = [] if args.secretos: with open(args.secretos, "r", encoding="utf-8", errors="replace") as fh: extra = [x.rstrip("\r\n") for x in fh if x.strip()] informe = {"formato": FORMATO, "generado": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "tokens": []} if args.buscar: hallados = buscar_en_ficheros(args.buscar) print(f"Tokens JWT encontrados: {len(hallados)}") for t, sitios in hallados.items(): try: r = analizar(t, extra) except ErrorJWT: continue r["ubicaciones"] = [{"fichero": f, "linea": n} for f, n in sitios] r["token"] = enmascarar(t) informe["tokens"].append(r) donde = ", ".join(f"{f}:{n}" for f, n in sitios[:3]) + (f" y {len(sitios) - 3} más" if len(sitios) > 3 else "") imprimir(r, f"● {enmascarar(t)} (huella {r['huella']}) en {donde}") if hallados: vivos = sum(1 for r in informe["tokens"] if isinstance((r["contenido"] or {}).get("exp"), (int, float)) and r["contenido"]["exp"] > time.time()) print(f"\nAún válidos: {vivos}. Los tokens en logs equivalen a contraseñas: revócalos y deja de registrar la cabecera Authorization.") else: token = args.token if args.token else sys.stdin.read() try: r = analizar(token, extra, args.secreto) except ErrorJWT as e: print(f"Error: {e}", file=sys.stderr) return 3 informe["tokens"].append(r) imprimir(r, "Token") print("\nCabecera:", json.dumps(r["cabecera"], ensure_ascii=False)) if r["contenido"] is not None: print("Contenido:", json.dumps(r["contenido"], ensure_ascii=False, indent=2)) if args.json: with open(args.json, "w", encoding="utf-8") as fh: json.dump(informe, fh, ensure_ascii=False, indent=2) print(f"\nInforme guardado en {args.json}") nivel = min((peor(r["hallazgos"]) for r in informe["tokens"]), default=3) return 2 if nivel == 0 else 1 if nivel == 1 else 0 if __name__ == "__main__": sys.exit(main())