dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/domain/hashchain.py

"""The tamper-evident chain (GoBD, FEC; GL-59): every posted journal carries chain_seq, prev_hash and row_hash.

row_hash = sha-256(prev_hash || the canonical journal, its lines and its split lines: the rows the balances sum). The lifecycle flags that legitimately move after
posting (status, reversal_status, reversal_reason, posted_at) are NOT part of it; every other stored column of the journal
and of its lines is, so changing any of them breaks verification at that journal, and the next link names it as its
predecessor.
"""
import hashlib
import json
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

GENESIS = "0" * 64

HEADER = ("journal_id", "company_code", "ledger_id", "journal_date", "period_id", "fiscal_year", "book", "entry_no", "source_module",
          "source_document", "event", "currency", "fx_rate", "memo", "total", "functional_total", "adjustment",
          "reverse_next_period", "ic_document_id", "recurring_id", "accrual_id", "allocation_run_id", "reval_run_id",
          "proposed_by", "proposed_sub", "approved_by", "approved_sub", "reverses_journal_id", "body_digest", "chain_seq",
          "migration_run_id")
LINE = ("line_no", "account_id", "purpose", "qualifier", "debit", "credit", "functional_debit", "functional_credit", "currency",
        "amount_currency", "bp_id", "tax_code", "cost_centre", "profit_centre", "partner_company", "movement_type",
        "against_document", "ic_reference", "asset_id", "line_memo", "odoo_line_id")
SPLIT = ("line_no", "split_no", "company_code", "ledger_id", "account_id", "period_id", "journal_date", "debit", "credit",
         "functional_debit", "functional_credit", "currency", "profit_centre", "segment", "cost_centre", "partner_company",
         "movement_type", "bp_id", "asset_id", "is_clearing", "adjustment")


def _v(x):
    if x is None:
        return ""
    if isinstance(x, Decimal):
        return "0" if x == 0 else format(x.normalize(), "f")
    if isinstance(x, (date, datetime)):
        return x.isoformat()
    if isinstance(x, bool):
        return "1" if x else "0"
    return str(x)


def canonical(journal: dict, lines: list, splits: list = ()) -> str:
    body = {"journal": {k: _v(journal.get(k)) for k in HEADER},
            "lines": [{k: _v(ln.get(k)) for k in LINE} for ln in sorted(lines, key=lambda r: r["line_no"])],
            "splits": [{k: _v(s.get(k)) for k in SPLIT} for s in sorted(splits, key=lambda r: (r["line_no"], r["split_no"]))]}
    return json.dumps(body, sort_keys=True, separators=(",", ":"))


def row_hash(prev_hash: str, journal: dict, lines: list, splits: list = ()) -> str:
    return hashlib.sha256((prev_hash + canonical(journal, lines, splits)).encode()).hexdigest()


def verify(chain: list, lines_of, splits_of=lambda jid: ()) -> Optional[dict]:
    """Walk a company's posted journals in chain order: `chain` is [journal row dicts by chain_seq]; `lines_of(journal_id)`
    gives a journal's lines. None when the chain holds, else {journal_id, chain_seq, why} for the FIRST broken link."""
    prev = GENESIS
    expected_seq = 1
    for j in chain:
        if j["chain_seq"] != expected_seq:
            return {"journal_id": j["journal_id"], "chain_seq": j["chain_seq"], "why": f"the chain skips to {j['chain_seq']} "
                                                                                      f"(expected {expected_seq})"}
        if j["prev_hash"] != prev:
            return {"journal_id": j["journal_id"], "chain_seq": j["chain_seq"], "why": "prev_hash is not the previous link's row_hash"}
        if row_hash(prev, j, lines_of(j["journal_id"]), splits_of(j["journal_id"])) != j["row_hash"]:
            return {"journal_id": j["journal_id"], "chain_seq": j["chain_seq"], "why": "the journal, its lines or its split lines changed after posting"}
        prev = j["row_hash"]
        expected_seq += 1
    return None