#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Verificador independente do pacote de integridade do Journey (formato journey-integridade/1). Uso: python verificador-integridade.py pacote.json [--tsa-ca cadeia_da_ACT.pem] [--saida pasta] O que ele faz, SEM depender do Journey nem da Baliza (só Python padrão): 1. Recomputa o hash de conteúdo de cada lote a partir das linhas brutas do pacote (sha256 das linhas unidas por \\n) e compara com o hash lacrado. 2. Confere a cadeia: hash_lote = sha256(hash_conteudo | hash_anterior). 3. Confere cada selo: raiz = sha256(hash_lote dos lotes do selo unidos por \\n) e hash_selo = sha256(raiz | hash_selo_anterior). 4. Extrai os carimbos de tempo RFC 3161 (.tsr) e as provas OpenTimestamps (.ots) e mostra como verificá-los com ferramentas públicas (openssl / ots). Se `openssl` estiver no PATH e você informar --tsa-ca, a verificação do carimbo é executada aqui mesmo. Interpretação: se tudo confere e o carimbo externo é válido, os dados existiam naquele instante e não foram alterados depois, nem pela empresa cliente nem pela Baliza. O verificador não prova que o agente capturou tudo (cobertura), só que o que foi capturado não mudou. """ import base64 import hashlib import json import os import shutil import subprocess import sys def sha(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def main() -> int: if len(sys.argv) < 2: print(__doc__) return 1 caminho = sys.argv[1] tsa_ca = None saida = os.path.dirname(os.path.abspath(caminho)) args = sys.argv[2:] for i, a in enumerate(args): if a == "--tsa-ca" and i + 1 < len(args): tsa_ca = args[i + 1] if a == "--saida" and i + 1 < len(args): saida = args[i + 1] os.makedirs(saida, exist_ok=True) with open(caminho, encoding="utf-8") as f: pkg = json.load(f) if pkg.get("formato") != "journey-integridade/1": print(f"AVISO: formato inesperado: {pkg.get('formato')}") # pacote de SELO (cópia WORM do GCS): um selo + seus lotes -> normaliza para o formato por pessoa if "selo" in pkg and "selos" not in pkg: s = dict(pkg["selo"]) s["lotes_no_selo"] = [l["hash_lote"] for l in sorted(pkg.get("lotes", []), key=lambda x: x["id"])] s.setdefault("ancora_status", "n/a (cópia WORM)") pkg["selos"] = [s] print(f"Pacote: usuário={pkg.get('usuario', '-')} período={pkg.get('periodo', '-')} gerado_em={pkg.get('gerado_em', '-')} schema={pkg.get('schema')}") tudo_ok = True lotes = sorted(pkg.get("lotes", []), key=lambda x: x["id"]) print(f"\n[1/3] Lotes: {len(lotes)}") for l in lotes: linhas = l.get("linhas") or [] h_conteudo = sha("\n".join(linhas)) if linhas else sha("") c_ok = h_conteudo == l["hash_conteudo"] h_lote = sha(l["hash_conteudo"] + "|" + (l.get("hash_anterior") or "")) ch_ok = h_lote == l["hash_lote"] n_ok = len(linhas) == int(l.get("n_registros", len(linhas))) status = "OK" if (c_ok and ch_ok and n_ok) else "FALHA" if status != "OK": tudo_ok = False print(f" lote #{l['id']} {l['dia']} {l['fonte']} disp={str(l['dispositivo_id'])[:8]} n={len(linhas)} " f"conteudo={'ok' if c_ok else 'DIVERGE'} cadeia={'ok' if ch_ok else 'DIVERGE'} contagem={'ok' if n_ok else 'DIVERGE'} -> {status}") selos = sorted(pkg.get("selos", []), key=lambda x: x["id"]) print(f"\n[2/3] Selos: {len(selos)}") for s in selos: raiz = sha("\n".join(s.get("lotes_no_selo") or [])) r_ok = raiz == s["raiz"] hs = sha(s["raiz"] + "|" + (s.get("hash_anterior") or "")) s_ok = hs == s["hash_selo"] # os lotes do pacote que pertencem ao selo precisam constar na raiz ids_no_pacote = [l["hash_lote"] for l in lotes if s["lote_id_ini"] <= l["id"] <= s["lote_id_fim"]] pert_ok = all(h in (s.get("lotes_no_selo") or []) for h in ids_no_pacote) status = "OK" if (r_ok and s_ok and pert_ok) else "FALHA" if status != "OK": tudo_ok = False print(f" selo #{s['id']} {s['criado_em']} lotes={s['n_lotes']} raiz={'ok' if r_ok else 'DIVERGE'} " f"cadeia={'ok' if s_ok else 'DIVERGE'} lotes_do_pacote_no_selo={'ok' if pert_ok else 'DIVERGE'} ancora={s.get('ancora_status')} -> {status}") print("\n[3/3] Âncoras externas") for s in selos: base = os.path.join(saida, f"selo-{s['id']}") if s.get("ancora_tsa_b64"): tsr = base + ".tsr" with open(tsr, "wb") as f: f.write(base64.b64decode(s["ancora_tsa_b64"])) print(f" selo #{s['id']}: carimbo RFC 3161 salvo em {tsr} (TSA: {s.get('ancora_tsa_url')})") print(f" verificar: openssl ts -verify -digest {s['hash_selo']} -in \"{tsr}\" -CAfile ") print(f" inspecionar: openssl ts -reply -in \"{tsr}\" -text") if tsa_ca and shutil.which("openssl"): r = subprocess.run(["openssl", "ts", "-verify", "-digest", s["hash_selo"], "-in", tsr, "-CAfile", tsa_ca], capture_output=True, text=True) ok = r.returncode == 0 and "Verification: OK" in (r.stdout + r.stderr) print(f" openssl: {'OK' if ok else 'FALHA'} {r.stdout.strip()} {r.stderr.strip()}") if not ok: tudo_ok = False else: print(f" selo #{s['id']}: sem carimbo RFC 3161") if s.get("ancora_icp_b64"): icp = base + ".icp.tsr" with open(icp, "wb") as f: f.write(base64.b64decode(s["ancora_icp_b64"])) print(f" carimbo ICP-Brasil salvo em {icp} (ACT: {s.get('ancora_icp_url')})") print(f" verificar: openssl ts -verify -digest {s['hash_selo']} -in \"{icp}\" -CAfile ") print(f" inspecionar: openssl ts -reply -in \"{icp}\" -text (politica 2.16.76.1.6.2 = ICP-Brasil)") if s.get("ancora_ots_b64"): ots = base + ".ots" with open(ots, "wb") as f: f.write(base64.b64decode(s["ancora_ots_b64"])) print(f" prova OpenTimestamps salva em {ots}: ots verify -d {s['hash_selo']} \"{ots}\" (pip install opentimestamps-client)") if s.get("ancora_gcs"): print(f" cópia WORM (retenção 5 anos, imutável): {s['ancora_gcs']}") print("\nRESULTADO:", "TUDO CONFERE" if tudo_ok else "HÁ DIVERGÊNCIAS (veja acima)") return 0 if tudo_ok else 2 if __name__ == "__main__": sys.exit(main())