#!/usr/bin/env python3 """ Implementación de referencia en Python del verificador abierto de El Registro de Jornada. Un solo fichero, solo biblioteca estándar (Python 3.9 o superior), sin red. Es INDEPENDIENTE de la del motor y de verificar.mjs: se escribió a partir de la especificación (docs/ESPECIFICACION-INTEGRIDAD.md) y debe dar el mismo veredicto (salida y códigos de problema) en todos los vectores de verificador/vectores/. Incluye Ed25519 (RFC 8032) en Python puro, lento pero suficiente para unos cientos de firmas. No sustituye a una librería criptográfica auditada: sirve para contrastar la especificación desde otro lenguaje. Uso: python verificar.py fichero.json lista de asientos o paquete (se detecta solo) python verificar.py recibo.txt --jwks jwks.json [--asiento asiento.json] python verificar.py paquete.json --jwks jwks.json Opción --json: salida legible por máquina. Salida: 0 todo cuadra, 1 hay problemas, 2 no se pudo leer o falta algo. No comprueba el sello de tiempo externo (RFC 3161); si el paquete lo lleva, lo avisa. Prueba integridad, no autoría ni veracidad. """ import base64 import hashlib import json import re import sys from datetime import datetime, timedelta, timezone from decimal import Decimal GENESIS = "0" * 64 FORMATO_PAQUETE = "erj-paquete-pericial" VERSION_PAQUETE = 1 TIPO_RECIBO = "erj-recibo+jws" TIPO_SELLO = "erj-sello-dia+jws" # ------------------------------------------------------------ canonización def _texto_json(s): """JSON.stringify de ECMAScript para una cadena: solo se escapan comillas, barra, controles y sustitutos sueltos.""" fuera = ['"'] cortos = {'"': '\\"', "\\": "\\\\", "\b": "\\b", "\f": "\\f", "\n": "\\n", "\r": "\\r", "\t": "\\t"} for c in s: o = ord(c) if c in cortos: fuera.append(cortos[c]) elif o < 0x20 or 0xD800 <= o <= 0xDFFF: fuera.append("\\u%04x" % o) else: fuera.append(c) fuera.append('"') return "".join(fuera) def _numero_js(v): """Number::toString de ECMAScript para un número JSON.""" if isinstance(v, int): if abs(v) <= 2**53: return str(v) v = float(v) if v == 0: return "0" signo = "-" if v < 0 else "" t = Decimal(repr(abs(v))).as_tuple() digitos = "".join(map(str, t.digits)).rstrip("0") or "0" k = len(digitos) n = t.exponent + len(t.digits) # valor = 0.digitos x 10^n if k <= n <= 21: cuerpo = digitos + "0" * (n - k) elif 0 < n <= 21: cuerpo = digitos[:n] + "." + digitos[n:] elif -6 < n <= 0: cuerpo = "0." + "0" * (-n) + digitos else: e = n - 1 mant = digitos if k == 1 else digitos[0] + "." + digitos[1:] cuerpo = "%se%s%d" % (mant, "+" if e >= 0 else "-", abs(e)) return signo + cuerpo def canonizar(v): """Claves ordenadas por unidades de código UTF-16, sin espacios; la misma salida que el motor.""" if v is None: return "null" if v is True: return "true" if v is False: return "false" if isinstance(v, str): return _texto_json(v) if isinstance(v, (int, float)): return _numero_js(v) if isinstance(v, list): return "[" + ",".join(canonizar(x) for x in v) + "]" claves = sorted(v.keys(), key=lambda k: k.encode("utf-16-be", "surrogatepass")) return "{" + ",".join(_texto_json(k) + ":" + canonizar(v[k]) for k in claves) + "}" def sha256(texto): return hashlib.sha256(texto.encode("utf-8", "surrogatepass")).hexdigest() # ----------------------------------------------------------------- Ed25519 _P = 2**255 - 19 _Q = 2**252 + 27742317777372353535851937790883648493 _D = -121665 * pow(121666, _P - 2, _P) % _P _I = pow(2, (_P - 1) // 4, _P) def _suma(a, b): A = (a[1] - a[0]) * (b[1] - b[0]) % _P B = (a[1] + a[0]) * (b[1] + b[0]) % _P C = 2 * a[3] * b[3] * _D % _P D = 2 * a[2] * b[2] % _P E, F, G, H = B - A, D - C, D + C, B + A return (E * F % _P, G * H % _P, F * G % _P, E * H % _P) def _multiplicar(s, p): r = (0, 1, 1, 0) while s > 0: if s & 1: r = _suma(r, p) p = _suma(p, p) s >>= 1 return r def _x_de(y, signo): x2 = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P x = pow(x2, (_P + 3) // 8, _P) if (x * x - x2) % _P != 0: x = x * _I % _P if (x * x - x2) % _P != 0: return None if x & 1 != signo: x = _P - x return x def _descomprimir(b): if len(b) != 32: return None n = int.from_bytes(b, "little") y, signo = n & ((1 << 255) - 1), n >> 255 if y >= _P: return None x = _x_de(y, signo) if x is None or (x == 0 and signo == 1): return None return (x, y, 1, x * y % _P) _BY = 4 * pow(5, _P - 2, _P) % _P _BASE = (_x_de(_BY, 0), _BY, 1, _x_de(_BY, 0) * _BY % _P) def _iguales(a, b): return (a[0] * b[2] - b[0] * a[2]) % _P == 0 and (a[1] * b[2] - b[1] * a[2]) % _P == 0 def ed25519_verifica(publica, mensaje, firma): """Verificación de RFC 8032 (con la comprobación de S < L).""" if len(publica) != 32 or len(firma) != 64: return False A = _descomprimir(publica) R = _descomprimir(firma[:32]) s = int.from_bytes(firma[32:], "little") if A is None or R is None or s >= _Q: return False h = int.from_bytes(hashlib.sha512(firma[:32] + publica + mensaje).digest(), "little") % _Q return _iguales(_multiplicar(s, _BASE), _suma(R, _multiplicar(h, A))) # --------------------------------------------------------------------- JWS def _de_b64url(t): if not re.fullmatch(r"[A-Za-z0-9_-]*", t): raise ValueError("base64url inválido") return base64.urlsafe_b64decode(t + "=" * ((4 - len(t) % 4) % 4)) def _claves(juego): r = [] if not isinstance(juego, dict): return r for k in juego.get("keys", []): if not isinstance(k, dict): continue jwk = k.get("jwk", k) if isinstance(k.get("kid"), str) and isinstance(jwk, dict): r.append((k["kid"], jwk)) return r def verificar_jws(jws, juego, tipo): partes = str(jws).strip().split(".") if len(partes) != 3: return {"ok": False, "motivo": "formato"} try: cab = json.loads(_de_b64url(partes[0]).decode("utf-8")) carga = json.loads(_de_b64url(partes[1]).decode("utf-8")) firma = _de_b64url(partes[2]) except Exception: return {"ok": False, "motivo": "formato"} if not isinstance(cab, dict) or not isinstance(carga, dict): return {"ok": False, "motivo": "formato"} if cab.get("alg") != "EdDSA": return {"ok": False, "motivo": "algoritmo"} if cab.get("typ") != tipo: return {"ok": False, "motivo": "tipo"} pub = next((j for kid, j in _claves(juego) if kid == cab.get("kid")), None) if pub is None: return {"ok": False, "motivo": "clave_desconocida", "kid": cab.get("kid")} try: ok = ed25519_verifica(_de_b64url(pub.get("x", "")), (partes[0] + "." + partes[1]).encode("ascii"), firma) except Exception: ok = False if ok: return {"ok": True, "carga": carga, "kid": cab.get("kid")} return {"ok": False, "motivo": "firma", "kid": cab.get("kid")} # ------------------------------------------------------------------ cadena def codigo_rotura(m): return m if m.startswith("cadena_") else "cadena_" + m def verificar_cadena(asientos, desde=None): previo = desde or {"secuencia": 0, "hash": GENESIS} roturas = [] ultimo = float("-inf") for a in asientos: a = a if isinstance(a, dict) else {} datos = {k: v for k, v in a.items() if k not in ("id", "secuencia", "hashPrevio", "hash", "registradoEn")} sec, prev, h, reg = a.get("secuencia"), a.get("hashPrevio"), a.get("hash"), a.get("registradoEn") if sec != previo["secuencia"] + 1: roturas.append((sec, a.get("id"), "secuencia_incorrecta", "se esperaba la secuencia %s y llegó la %s" % (previo["secuencia"] + 1, sec))) if prev != previo["hash"]: roturas.append((sec, a.get("id"), "cadena_discontinua", "su hashPrevio no es el resumen del asiento anterior")) if sha256("%s|%s" % (prev, canonizar(datos))) != h: roturas.append((sec, a.get("id"), "hash_no_corresponde", "su contenido no produce su propio resumen")) if isinstance(reg, (int, float)) and not isinstance(reg, bool): if reg < ultimo: roturas.append((sec, a.get("id"), "orden_temporal", "consta registrado antes que un asiento de secuencia menor")) ultimo = max(ultimo, reg) previo = {"secuencia": sec, "hash": h} return roturas, previo def _fecha_utc(ms): return (datetime(1970, 1, 1, tzinfo=timezone.utc) + timedelta(milliseconds=ms)).strftime("%Y-%m-%d") def _dia_siguiente(f): return (datetime.strptime(f, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d") def huella_del_dia(empresa, fecha, del_dia): o = sorted(del_dia, key=lambda a: a["secuencia"]) return sha256(canonizar({"v": 1, "empresaId": empresa, "fecha": fecha, "previo": o[0]["hashPrevio"] if o else None, "cadena": [[a["secuencia"], a["hash"]] for a in o]})) def _contrastar(claves_paquete, jwks, nombre, p, avisos): publicadas = dict(_claves(jwks)) for kid, jwk in _claves(claves_paquete): if kid not in publicadas: avisos.append("la clave «%s» (%s) no está en el JWKS aportado" % (kid, nombre)) elif publicadas[kid].get("x") != jwk.get("x"): p("clave_no_coincide_jwks", "la clave «%s» (%s) del paquete NO es la publicada" % (kid, nombre)) def _resultado(tipo, problemas, avisos, huella=None): return {"tipo": tipo, "valido": not problemas, "problemas": [{"codigo": c, "detalle": d} for c, d in problemas], "avisos": avisos, "huella": huella} # ----------------------------------------------------------------- paquete def verificar_paquete(texto, jwks=None): problemas, avisos = [], [] def p(c, d): problemas.append((c, d)) try: paquete = json.loads(texto) except Exception as e: p("no_es_json", "el archivo no es JSON válido: %s" % e) return _resultado("paquete", problemas, avisos) if not isinstance(paquete, dict) or paquete.get("formato") != FORMATO_PAQUETE: p("formato", "no es un paquete pericial") return _resultado("paquete", problemas, avisos) if paquete.get("version") != VERSION_PAQUETE: p("version", "versión no soportada") return _resultado("paquete", problemas, avisos) c = paquete.get("contenido") if not isinstance(c, dict) or not isinstance(c.get("asientos"), list) or not c.get("manifiesto") or not c.get("tramo"): p("estructura", "faltan partes obligatorias del paquete") return _resultado("paquete", problemas, avisos) # 1. Forma canónica byte a byte y huella global. if texto != canonizar(paquete): p("no_canonico", "el archivo no es byte a byte el generado") huella = sha256(canonizar(c)) if huella != paquete.get("huella"): p("huella_global", "la huella global no corresponde al contenido") # 2. Huellas de cada parte y recuentos. m = c["manifiesto"] sellos = c.get("sellos") or [] recibos = c.get("recibos") or [] firmados = c.get("sellosFirmados") or [] partes = { "asientos": canonizar(c["asientos"]), "sellos": canonizar(sellos), "recibos": canonizar({"recibos": recibos, "clavesPublicas": c.get("clavesPublicas") or {"keys": []}}), "informeTxt": str((c.get("informe") or {}).get("txt", "")), "informeHtml": str((c.get("informe") or {}).get("html", "")), "instrucciones": str(c.get("instrucciones", "")), "incidencias": canonizar(c.get("incidencias") or []), } huellas, recuentos = m.get("huellas") or {}, m.get("recuentos") or {} con_firmados = "sellosFirmados" in c or "sellosFirmados" in huellas or "sellosFirmados" in recuentos if con_firmados: partes["sellosFirmados"] = canonizar({"sellosFirmados": firmados, "clavesSellos": c.get("clavesSellos") or {"keys": []}}) for nombre, valor in partes.items(): if huellas.get(nombre) != sha256(valor): p("huella_parte", "la huella de «%s» no corresponde al manifiesto" % nombre) if con_firmados and recuentos.get("sellosFirmados") != len(firmados): p("recuento", "el recuento de sellos firmados no cuadra") if recuentos.get("asientos") != len(c["asientos"]): p("recuento", "el recuento de asientos no cuadra") if recuentos.get("sellos") != len(sellos): p("recuento", "el recuento de sellos no cuadra") if recuentos.get("recibos") != len(recibos): p("recuento", "el recuento de recibos no cuadra") # 3. Cadena del tramo. asientos, t = c["asientos"], c["tramo"] if not asientos: p("sin_asientos", "el paquete no contiene asientos") return _resultado("paquete", problemas, avisos, huella) primero, ultimo = asientos[0], asientos[-1] if primero.get("secuencia") != t.get("desdeSecuencia"): p("tramo", "la secuencia inicial declarada no es la del primer asiento") if ultimo.get("secuencia") != t.get("hastaSecuencia"): p("tramo", "la secuencia final declarada no es la del último asiento") if primero.get("hashPrevio") != t.get("hashCabeza"): p("tramo", "el hash de cabeza declarado no es el hashPrevio del primer asiento") if ultimo.get("hash") != t.get("hashCola"): p("tramo", "el hash de cola declarado no es el del último asiento") desde = {"secuencia": t["desdeSecuencia"] - 1, "hash": t["hashCabeza"]} roturas, _ = verificar_cadena(asientos, desde) for sec, id_, mot, det in roturas: p(codigo_rotura(mot), "secuencia %s (%s): %s" % (sec, id_, det)) if desde["secuencia"] == 0 and desde["hash"] != GENESIS: p("tramo", "el tramo empieza en la secuencia 1 pero su cabeza no es el génesis") emp = (c.get("empresa") or {}).get("id") if emp is not None: for a in asientos: if a.get("empresaId") != emp: p("empresa", "el asiento %s es de otra empresa" % a.get("id")) # 4. Sellos diarios encadenados. def hash_en(sec): if sec == desde["secuencia"]: return desde["hash"] i = sec - t["desdeSecuencia"] return asientos[i]["hash"] if 0 <= i < len(asientos) and asientos[i]["secuencia"] == sec else None previo = None for s in sellos: datos = {k: v for k, v in s.items() if k not in ("hash", "selladoEn")} if sha256(canonizar(datos)) != s.get("hash"): p("sello_alterado", "el sello del %s no produce su propio resumen" % s.get("fecha")) if emp is not None and s.get("empresaId") != emp: p("empresa", "el sello del %s es de otra empresa" % s.get("fecha")) if previo: if s.get("selloPrevio") != previo["hash"]: p("serie_discontinua", "el sello del %s no enlaza con el del %s" % (s.get("fecha"), previo["fecha"])) if s.get("fecha") != _dia_siguiente(previo["fecha"]): p("dia_sin_sellar", "entre el %s y el %s hay días sin sellar" % (previo["fecha"], s.get("fecha"))) hasta = s["hastaSecuencia"] previo_hasta = previo["hastaSecuencia"] if previo else hasta - s["asientos"] h_cierre, h_previo = hash_en(hasta), hash_en(previo_hasta) if h_cierre is None or h_previo is None: avisos.append("el sello del %s cierra fuera del tramo aportado" % s.get("fecha")) else: if h_cierre != s.get("hashCierre"): p("cierre_no_corresponde", "el sello del %s declara un cierre que no es el del asiento %s" % (s.get("fecha"), hasta)) if hasta - previo_hasta != s["asientos"]: p("recuento_sello", "el sello del %s declara %s asientos y el tramo tiene %s" % (s.get("fecha"), s["asientos"], hasta - previo_hasta)) previo = s # 5. Recibos firmados. por_id = {a.get("id"): a for a in asientos} for r in recibos: v = verificar_jws(r, c.get("clavesPublicas"), TIPO_RECIBO) if not v["ok"]: p("recibo_invalido", "un recibo no es válido: %s" % v["motivo"]) continue k = v["carga"] a = por_id.get(k.get("id")) if not a or k.get("emp") != a.get("empresaId") or k.get("seq") != a.get("secuencia") or k.get("hash") != a.get("hash") or k.get("prev") != a.get("hashPrevio"): p("recibo_no_coincide", "el recibo del asiento %s no coincide con ningún asiento del tramo" % k.get("id")) # 6. Sellos diarios firmados. vistas = set() con_tiempo = 0 for s in firmados: if not isinstance(s, dict) or not isinstance(s.get("jws"), str): p("sello_firmado_formato", "un sello firmado no tiene la forma esperada") continue et = "el sello firmado del %s" % s.get("fecha") if s.get("fecha") in vistas: p("sello_firmado_repetido", "hay dos sellos firmados del %s" % s.get("fecha")) vistas.add(s.get("fecha")) if emp is not None and s.get("empresaId") != emp: p("empresa", "%s es de otra empresa" % et) v = verificar_jws(s["jws"], c.get("clavesSellos"), TIPO_SELLO) if not v["ok"]: p("sello_firmado_invalido", "%s no es válido: %s" % (et, v["motivo"])) continue k = v["carga"] if v["kid"] != s.get("kid"): p("sello_firmado_no_coincide", "%s: la clave del índice no es la de la firma" % et) if k.get("v") != 1 or k.get("iss") != "erj": p("sello_firmado_no_coincide", "%s: versión o emisor no reconocidos" % et) if k.get("emp") != s.get("empresaId") or k.get("fecha") != s.get("fecha") or k.get("n") != s.get("asientos") or k.get("seq") != s.get("hastaSecuencia") or k.get("huella") != s.get("huella"): p("sello_firmado_no_coincide", "%s: los datos firmados no coinciden con los del índice" % et) if emp is not None and k.get("emp") != emp: p("empresa", "%s firma otra empresa" % et) per = m.get("periodo") if per and per["desde"] <= s.get("fecha", "") <= per["hasta"]: del_dia = [a for a in asientos if _fecha_utc(a["registradoEn"]) == s["fecha"]] seq = max([a["secuencia"] for a in del_dia] or [0]) if len(del_dia) != k.get("n"): p("sello_firmado_recuento", "%s declara %s asientos y el paquete tiene %s de ese día" % (et, k.get("n"), len(del_dia))) elif seq != k.get("seq"): p("sello_firmado_recuento", "%s cierra en la secuencia %s y el paquete en la %s" % (et, k.get("seq"), seq)) if huella_del_dia(k.get("emp"), k.get("fecha"), del_dia) != k.get("huella"): p("sello_firmado_huella", "%s no coincide con la cadena de asientos de ese día" % et) else: avisos.append("%s cae fuera del periodo: solo se comprobó su firma" % et) if "selloTiempo" in s: con_tiempo += 1 if con_tiempo: avisos.append("%d sello(s) de tiempo externo (RFC 3161): esta implementación NO los comprueba" % con_tiempo) # 7. Contraste con el JWKS publicado. if jwks: _contrastar(c.get("clavesPublicas"), jwks, "recibos", p, avisos) _contrastar(c.get("clavesSellos"), jwks, "sellos diarios", p, avisos) elif recibos or firmados: avisos.append("las firmas se comprobaron con las claves del propio paquete; aporte el JWKS publicado") return _resultado("paquete", problemas, avisos, huella) # ------------------------------------------------------ lista, recibo, todo def verificar_recibo(recibo, jwks, asiento=None): problemas, avisos = [], [] v = verificar_jws(recibo, jwks, TIPO_RECIBO) if not v["ok"]: problemas.append(("recibo_invalido", "el recibo no es válido: %s" % v["motivo"])) return _resultado("recibo", problemas, avisos) if asiento: datos = {k: x for k, x in asiento.items() if k not in ("id", "secuencia", "hashPrevio", "hash", "registradoEn")} k = v["carga"] if sha256("%s|%s" % (asiento.get("hashPrevio"), canonizar(datos))) != asiento.get("hash"): problemas.append(("asiento_alterado", "el asiento aportado no produce su propio resumen")) if k.get("id") != asiento.get("id") or k.get("emp") != asiento.get("empresaId") or k.get("seq") != asiento.get("secuencia") or k.get("hash") != asiento.get("hash") or k.get("prev") != asiento.get("hashPrevio"): problemas.append(("recibo_no_coincide", "el recibo no corresponde al asiento aportado")) else: avisos.append("la firma es válida, pero no se ha comparado con ningún asiento") return _resultado("recibo", problemas, avisos) def verificar(texto, jwks=None, asiento=None): limpio = texto[1:] if texto.startswith("") else texto t = limpio.strip() if re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]*", t): if not jwks: return {"tipo": "recibo", "valido": False, "necesita": "jwks", "problemas": [{"codigo": "falta_jwks", "detalle": "hacen falta las claves públicas (JWKS)"}], "avisos": [], "huella": None} return verificar_recibo(t, jwks, asiento) try: j = json.loads(limpio) except Exception as e: return _resultado("desconocido", [("no_es_json", "no es un JSON ni un recibo firmado: %s" % e)], []) if isinstance(j, list): roturas, _ = verificar_cadena(j) avisos = [] if j and isinstance(j[0], dict) and j[0].get("secuencia") != 1: avisos.append("la lista no empieza en la secuencia 1: es un tramo") return _resultado("cadena", [(codigo_rotura(m), "secuencia %s (%s): %s" % (s, i, d)) for s, i, m, d in roturas], avisos) if isinstance(j, dict) and j.get("formato") == FORMATO_PAQUETE: return verificar_paquete(limpio, jwks) if isinstance(j, dict) and isinstance(j.get("keys"), list): return _resultado("desconocido", [("es_un_jwks", "son claves públicas (JWKS), no algo que verificar")], []) return _resultado("desconocido", [("formato", "no se reconoce")], []) def _principal(args): def valor(n): return args[args.index(n) + 1] if n in args and args.index(n) + 1 < len(args) else None ruta = next((a for i, a in enumerate(args) if not a.startswith("--") and (i == 0 or args[i - 1] not in ("--jwks", "--asiento"))), None) if not ruta: print("Uso: python verificar.py fichero.json|recibo.txt [--jwks jwks.json] [--asiento asiento.json] [--json]", file=sys.stderr) return 2 try: with open(ruta, encoding="utf-8", newline="") as f: texto = f.read() jwks = json.load(open(valor("--jwks"), encoding="utf-8")) if "--jwks" in args else None asiento = json.load(open(valor("--asiento"), encoding="utf-8")) if "--asiento" in args else None except (OSError, ValueError, TypeError) as e: print("No se puede leer: %s" % e, file=sys.stderr) return 2 r = verificar(texto, jwks, asiento) if r.get("necesita") == "jwks": print("Para comprobar un recibo hacen falta las claves públicas: --jwks jwks.json", file=sys.stderr) return 2 if r["tipo"] == "desconocido" and not r["valido"] and r["problemas"][0]["codigo"] != "es_un_jwks": print("No se puede interpretar %s: %s" % (ruta, r["problemas"][0]["detalle"]), file=sys.stderr) return 2 if "--json" in args: sys.stdout.reconfigure(encoding="utf-8") print(json.dumps(r, ensure_ascii=False, indent=2)) else: sys.stdout.reconfigure(encoding="utf-8") for a in r["avisos"]: print(" aviso: " + a) print(("%s VERIFICADO: todo cuadra. Prueba integridad, no autoría." if r["valido"] else "%s NO VERIFICADO.") % r["tipo"].upper()) for x in r["problemas"]: print(" · %s: %s" % (x["codigo"], x["detalle"])) return 0 if r["valido"] else 1 if __name__ == "__main__": sys.exit(_principal(sys.argv[1:]))