#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""verify_ledger.py - verify a DeepSweep agent evidence LEDGER offline, Python standard library only.

    python3 verify_ledger.py ledger.jsonl ledger-sig.jsonl ledger.pub.pem
    python3 verify_ledger.py ledger.jsonl ledger-sig.jsonl ledger.pub.pem --tree-head ledger-head.json
    python3 verify_ledger.py ledger.jsonl ledger-sig.jsonl ledger.pub.pem --bundle evidence-bundle.json

Keep verify_bundle.py in the same folder: it supplies the SHA-256 / RFC 6962 Merkle code, Ed25519
(RFC 8032) and the bundle check, so each of those exists once in this kit.

Exit 0 = VERIFIED: every entry chains from genesis and carries a signature that verifies under the pinned key.
Exit 3 = UNSIGNED: the chain is internally consistent, but nothing attests to it. This is never a pass.
Exit 4 = NOT VERIFIED: a check failed; the reason is printed (edited, reordered, truncated, forged,
         untrusted-key, signature-gap, head-refused, malformed-signature, malformed-ledger).
Exit 2 = an input could not be read: a file is missing or is not UTF-8 text, or the public key is not
         an Ed25519 public key. Every other input, however malformed, ends in exit 0, 3 or 4.

The checks run in a fixed order and the first failure proven names the verdict, because a ledger can be
wrong in several ways at once and the most specific cause is the useful one:

  0 a ledger line is not one JSON entry with exactly
    seq, prevHash, occurredAt, kind, payload, entryHash -> malformed-ledger
  1 sidecar unreadable                    -> malformed-signature
  2 entries out of seq order, but chain
    when sorted                           -> reordered
  3 hash chain broken                     -> edited
  4 more signatures than entries          -> truncated
  5 a supplied tree head or bundle cannot
    be read, or is refused                -> head-refused
  6 it attests more entries than exist    -> truncated
  7 the tree head's root does not
    reproduce over its first entries      -> edited
    (a bundle is compared by entry count only: its records are not the ledger's bytes)
  8 no signatures at all                  -> unsigned (exit 3)
  9 per entry: signature missing          -> signature-gap
               covers a different hash    -> edited
               signed by another key      -> untrusted-key
               signature does not verify  -> forged
 10 otherwise                             -> verified

JSON is read strictly, exactly as the in-browser checker reads it, so the two reach the same verdict:
NaN and Infinity, numbers too large to be finite, integers beyond +/-(2^53 - 1), text that is not valid
Unicode, an object that names the same field twice, a byte-order mark and nesting deeper than 64 levels
are all refused. Canonical JSON is RFC 8785,
including its number format, so 3 and 3.0 are the same value and hash the same.

WHAT A PASS DOES AND DOES NOT TELL YOU. A pass shows that every entry is signed by the pinned key, and
that nobody who lacks that key changed an entry, moved one, or took one out of the middle since it was
signed. It does NOT show that the key holder left the ledger unchanged: whoever holds the signing key can
rewrite and re-sign the whole ledger and it will pass. So will a new tree head and bundle signed to match.
Only a tree head you received before the rewrite, or one with an independent timestamp (checked
separately; this script does not check timestamps), catches that. A bundle you received before catches it only if the rewrite
left fewer entries, because a bundle is compared by count. Nor does a pass show the entries are true,
that every action was recorded, when they were really written (timestamps are the recorder's claim), or
who holds the key. Cutting entries off the END of both files is only detectable against a tree head or a
bundle signed earlier (--tree-head, --bundle), because the cut ledger has fewer entries: without one, a
shortened ledger verifies.
"""
import base64
import hashlib
import json
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from verify_bundle import (  # noqa: E402 - the sibling file is part of this kit
    ed25519_verify,
    hash_leaf,
    merkle_root,
    verify_bundle,
)

GENESIS_PREV = "0" * 64
HEX = re.compile(r"[0-9a-f]{64}")
B64 = re.compile(r"[A-Za-z0-9+/]*={0,2}")
SURROGATE = re.compile("[\ud800-\udfff]")
SPKI_ED25519_PREFIX = bytes.fromhex("302a300506032b6570032100")
ENTRY_FIELDS = frozenset(("seq", "prevHash", "occurredAt", "kind", "payload", "entryHash"))
MAX_SAFE_INTEGER = 2**53 - 1
MAX_DEPTH = 64

EXIT = {"verified": 0, "unsigned": 3, "unreadable": 2}


class Malformed(ValueError):
    pass


class Unreadable(object):
    """A tree head or bundle whose text could not be read. Refused at step 5, not before."""

    def __init__(self, detail):
        self.detail = detail


# ----------------------------------------------------------------- strict JSON


def _reject_constant(name):
    raise Malformed("non-standard JSON constant %s" % name)


def _parse_int(token):
    value = int(token)
    if abs(value) > MAX_SAFE_INTEGER:
        raise Malformed("an integer outside +/-(2^53 - 1)")
    return value


def _parse_float(token):
    value = float(token)
    if value in (float("inf"), float("-inf")):
        raise Malformed("a number too large to be finite")
    return value


def _object(pairs):
    """RFC 8785 and I-JSON require unique names. Keeping the last of two would let a raw reader see a
    different value from the one that was hashed and signed."""
    out = {}
    for k, v in pairs:
        if k in out:
            raise Malformed("an object names the same field twice")
        out[k] = v
    return out


def _check_values(value):
    """Refuse text that is not valid Unicode and nesting deeper than MAX_DEPTH. Iterative on purpose."""
    stack = [(value, 1)]
    while stack:
        v, depth = stack.pop()
        if isinstance(v, str):
            if SURROGATE.search(v):
                raise Malformed("text that is not valid Unicode (a lone surrogate)")
        elif isinstance(v, (list, dict)):
            if depth > MAX_DEPTH:
                raise Malformed("JSON nested deeper than %d levels" % MAX_DEPTH)
            if isinstance(v, dict):
                for k, x in v.items():
                    if SURROGATE.search(k):
                        raise Malformed("text that is not valid Unicode (a lone surrogate)")
                    stack.append((x, depth + 1))
            else:
                for x in v:
                    stack.append((x, depth + 1))


def loads(text):
    """Strict JSON: raises Malformed for anything the in-browser checker would not read the same way."""
    try:
        value = json.loads(
            text,
            parse_constant=_reject_constant,
            parse_int=_parse_int,
            parse_float=_parse_float,
            object_pairs_hook=_object,
        )
    except Malformed:
        raise
    except (ValueError, RecursionError) as exc:
        raise Malformed("not valid JSON (%s)" % type(exc).__name__)
    _check_values(value)
    return value


def _es_number(v):
    """RFC 8785 number: the ECMAScript Number-to-String form of a finite double."""
    if v != v or v in (float("inf"), float("-inf")):
        raise Malformed("a number that is not finite")
    if v == 0:
        return "0"
    if v.is_integer() and abs(v) <= MAX_SAFE_INTEGER:
        return str(int(v))
    text = repr(v)
    sign = "-" if text.startswith("-") else ""
    text = text.lstrip("-")
    mantissa, _, exp = text.partition("e")
    whole, _, frac = mantissa.partition(".")
    raw = whole + frac
    lead = len(raw) - len(raw.lstrip("0"))
    digits = raw.lstrip("0").rstrip("0")
    k = len(digits)
    n = len(whole) - lead + (int(exp) if exp else 0)
    if k <= n <= 21:
        out = digits + "0" * (n - k)
    elif 0 < n <= 21:
        out = digits[:n] + "." + digits[n:]
    elif -6 < n <= 0:
        out = "0." + "0" * (-n) + digits
    else:
        e = n - 1
        out = digits[0] + ("." + digits[1:] if k > 1 else "") + "e" + ("+" if e >= 0 else "-") + str(abs(e))
    return sign + out


def canonical(v):
    """RFC 8785 (JCS). Keys sort by UTF-16 code units; numbers use the ECMAScript form."""
    if v is None:
        return "null"
    if v is True:
        return "true"
    if v is False:
        return "false"
    if isinstance(v, int):
        return str(v)
    if isinstance(v, float):
        return _es_number(v)
    if isinstance(v, str):
        return json.dumps(v, ensure_ascii=False)
    if isinstance(v, list):
        return "[" + ",".join(canonical(x) for x in v) + "]"
    if isinstance(v, dict):
        keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
        return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + canonical(v[k]) for k in keys) + "}"
    raise Malformed("a value canonical JSON cannot hold")


def _sha256_hex(data):
    return hashlib.sha256(data).hexdigest()


def _is_number(v):
    return isinstance(v, (int, float)) and not isinstance(v, bool)


def _is_int(v):
    return isinstance(v, int) and not isinstance(v, bool)


def _is_hex(v):
    return isinstance(v, str) and HEX.fullmatch(v) is not None


def short_id(text):
    """An author-supplied key id, cut to a readable length before it reaches a message."""
    return text if len(text) <= 32 else text[:32] + "..."


def _is_blank(line):
    return line.strip(" \t\r") == ""


# ----------------------------------------------------------------- inputs


def load_key(text):
    """A PEM (or bare base64 DER) Ed25519 public key -> (raw 32 bytes, dsk_ key id). Strict base64."""
    t = text.strip(" \t\r\n")
    if "BEGIN PUBLIC KEY" in t:
        parts = [line.strip(" \t\r") for line in t.split("\n")]
        b64 = "".join(p for p in parts if p and not p.startswith("-----"))
    else:
        b64 = t
    der = strict_b64(b64)
    if der is None or len(der) != 44 or der[:12] != SPKI_ED25519_PREFIX:
        raise ValueError("not an Ed25519 public key")
    key_id = "dsk_" + _sha256_hex(base64.b64encode(der))[:16]
    return der[12:], key_id


def parse_ledger(text):
    """One JSON entry per line; blank lines ignored. Any malformed line refuses the whole ledger."""
    entries = []
    for line in text.split("\n"):
        if _is_blank(line):
            continue
        e = loads(line)
        if (
            not isinstance(e, dict)
            or set(e) != ENTRY_FIELDS
            or not _is_number(e["seq"])
            or not isinstance(e["prevHash"], str)
            or not isinstance(e["occurredAt"], str)
            or not isinstance(e["kind"], str)
            or not isinstance(e["entryHash"], str)
            or not isinstance(e["payload"], dict)
        ):
            raise Malformed("a ledger line is not exactly a {seq, prevHash, occurredAt, kind, payload, entryHash} entry")
        entries.append(e)
    return entries


def parse_signatures(text):
    """Returns None when the sidecar is malformed: the ladder refuses rather than reading part of it."""
    out = []
    for line in text.split("\n"):
        if _is_blank(line):
            continue
        try:
            s = loads(line)
        except Malformed:
            return None
        if not isinstance(s, dict):
            return None
        sv, seq = s.get("schemaVersion"), s.get("seq")
        if (
            not (_is_number(sv) and sv == 1)
            or not (_is_number(seq) and float(seq).is_integer() and seq >= 0)
            or not isinstance(s.get("entryHash"), str)
            or not isinstance(s.get("keyId"), str)
            or not isinstance(s.get("signature"), str)
        ):
            return None
        out.append(s)
    return out


def read_optional(text):
    """A tree head or bundle text -> the parsed value, or Unreadable (refused later, at step 5)."""
    if text is None:
        return None
    try:
        return loads(text)
    except Malformed as exc:
        return Unreadable(str(exc))


# ----------------------------------------------------------------- the checks


def entry_hash(e):
    body = {k: e[k] for k in ("seq", "prevHash", "occurredAt", "kind", "payload")}
    return _sha256_hex(canonical(body).encode("utf-8"))


def chain_ok(entries):
    prev = GENESIS_PREV
    for i, e in enumerate(entries):
        if e["seq"] != i or e["prevHash"] != prev or entry_hash(e) != e["entryHash"]:
            return False
        prev = e["entryHash"]
    return True


def ledger_root(entries):
    leaves = []
    for e in entries:
        if not _is_hex(e["entryHash"]):
            return None
        leaves.append(hash_leaf(bytes.fromhex(e["entryHash"])))
    return merkle_root(leaves)


def strict_b64(s):
    if not isinstance(s, str) or len(s) % 4 != 0 or B64.fullmatch(s) is None:
        return None
    try:
        return base64.b64decode(s, validate=True)
    except ValueError:
        return None


def _head_shape_ok(sth):
    if not isinstance(sth, dict):
        return False
    head = sth.get("treeHead")
    return (
        isinstance(head, dict)
        and _is_number(head.get("schemaVersion"))
        and head.get("schemaVersion") == 1
        and _is_int(head.get("treeSize"))
        and head.get("treeSize") >= 0
        and _is_hex(head.get("rootHash"))
        and isinstance(head.get("signedAt"), str)
        and isinstance(head.get("logId"), str)
        and isinstance(sth.get("signature"), str)
        and isinstance(sth.get("keyId"), str)
    )


def check_tree_head(sth, pinned_raw, pinned_key_id):
    """Returns (ok, detail) for a signed tree head envelope {treeHead, signature, keyId}."""
    if not isinstance(sth, dict):
        return False, "tree head is not an object"
    if not _head_shape_ok(sth):
        return False, "malformed signed tree head envelope"
    head, sig, kid = sth["treeHead"], sth["signature"], sth["keyId"]
    if kid != pinned_key_id:
        return False, "tree head keyId %s is not the pinned key (%s)" % (short_id(kid), pinned_key_id)
    if head["logId"] != kid:
        return False, "treeHead.logId does not match the signing keyId"
    raw = strict_b64(sig)
    if raw is None or not ed25519_verify(pinned_raw, canonical(head).encode("utf-8"), raw):
        return False, "tree head signature does not verify under %s" % kid
    return True, "tree head signed by %s" % kid


def _numbers_are_safe_integers(value):
    """True when every number inside value is an integer within +/-(2^53 - 1)."""
    stack = [value]
    while stack:
        v = stack.pop()
        if _is_number(v):
            if isinstance(v, float) and not (v.is_integer() and abs(v) <= MAX_SAFE_INTEGER):
                return False
        elif isinstance(v, dict):
            stack.extend(v.values())
        elif isinstance(v, list):
            stack.extend(v)
    return True


def _proof_ok(items):
    return isinstance(items, list) and all(_is_hex(p) for p in items)


def bundle_problem(bundle):
    """What the in-browser checker refuses in a bundle that verify_bundle.py would read more leniently.

    verify_bundle.py is vendored byte for byte, so its leniencies stay in it. Before this kit relies on
    it, the bundle must carry strict base64, real integers (not true/false) for sizes and indexes, exact
    lowercase hex digests in every proof, and only safe integers in the signed content. Returns a reason,
    or None."""
    if not isinstance(bundle, dict):
        return None  # verify_bundle refuses it as an envelope
    sth = bundle.get("signedTreeHead")
    if sth is not None:
        if not _head_shape_ok(sth):
            return "malformed signed tree head envelope"
        if strict_b64(sth["signature"]) is None:
            return "the tree head signature is not strict base64"
        if not _numbers_are_safe_integers(sth["treeHead"]):
            return "the tree head holds a number that is not a safe integer"
    records = bundle.get("records")
    if isinstance(records, list) and not _numbers_are_safe_integers(records):
        return "a record holds a number that is not a safe integer"
    inclusion = bundle.get("inclusion")
    if isinstance(inclusion, list):
        for inc in inclusion:
            if isinstance(inc, dict):
                if isinstance(inc.get("leafIndex"), bool) or isinstance(inc.get("treeSize"), bool):
                    return "an inclusion proof gives true or false where a number belongs"
                if isinstance(inc.get("leafHash"), str) and not _is_hex(inc["leafHash"]):
                    return "an inclusion proof names a leaf hash that is not a digest"
                if isinstance(inc.get("proof"), list) and not _proof_ok(inc["proof"]):
                    return "an inclusion proof holds a step that is not a digest"
        indexes = sorted(inc.get("leafIndex") for inc in inclusion if isinstance(inc, dict) and _is_int(inc.get("leafIndex")))
        if isinstance(records, list) and indexes != list(range(len(records))):
            return "the inclusion proofs do not cover each record exactly once"
    cons = bundle.get("consistency")
    if isinstance(cons, list):
        for c in cons:
            if isinstance(c, dict):
                if isinstance(c.get("firstSize"), bool) or isinstance(c.get("secondSize"), bool):
                    return "a consistency proof gives true or false where a number belongs"
                if isinstance(c.get("firstRoot"), str) and not _is_hex(c["firstRoot"]):
                    return "a consistency proof names a root that is not a digest"
                if isinstance(c.get("proof"), list) and not _proof_ok(c["proof"]):
                    return "a consistency proof holds a step that is not a digest"
    return None


def verify_ledger(entries, signatures, pinned_raw, pinned_key_id, tree_head=None, bundle=None):
    """The ladder. Returns (status, detail). tree_head and bundle are parsed values, or Unreadable."""
    n = len(entries)
    if signatures is None:
        return "malformed-signature", "the signature file is malformed; refusing to verify against unreadable evidence"
    out_of_order = any(e["seq"] != i for i, e in enumerate(entries))
    if out_of_order and chain_ok(sorted(entries, key=lambda e: e["seq"])):
        return "reordered", "entries are not in seq order but chain when sorted: history was reordered (%d entries)" % n
    if not chain_ok(entries):
        return "edited", "the hash chain does not verify over %d entries: an entry was edited or removed" % n
    if len(signatures) > n:
        return "truncated", "%d signatures cover only %d entries: the ledger was cut beneath its own signatures" % (
            len(signatures),
            n,
        )
    if tree_head is not None:
        if isinstance(tree_head, Unreadable):
            return "head-refused", "the ledger tree head cannot be read (%s)" % tree_head.detail
        ok, detail = check_tree_head(tree_head, pinned_raw, pinned_key_id)
        if not ok:
            return "head-refused", "the ledger tree head was refused (%s)" % detail
        size = tree_head["treeHead"]["treeSize"]
        if size > n:
            return "truncated", "the signed tree head attests %d entries but only %d remain" % (size, n)
        if ledger_root(entries[:size]) != tree_head["treeHead"]["rootHash"]:
            return "edited", "the signed tree head root does not reproduce over the first %d entries" % size
    if bundle is not None:
        if isinstance(bundle, Unreadable):
            return "head-refused", "the evidence bundle cannot be read (%s)" % bundle.detail
        problem = bundle_problem(bundle)
        if problem is not None:
            return "head-refused", "the evidence bundle was refused (%s)" % problem
        try:
            ok, findings = verify_bundle(bundle, pinned_raw, pinned_key_id)
        except Exception as exc:  # the vendored checker can raise on shapes it does not expect
            return "head-refused", "the evidence bundle could not be checked (%s)" % type(exc).__name__
        if not ok:
            failed = ", ".join(c for c, good, _ in findings if not good)
            return "head-refused", "the evidence bundle was refused (failed: %s)" % failed
        size = bundle["signedTreeHead"]["treeHead"]["treeSize"]
        if size > n:
            return "truncated", "the signed bundle attests %d records but only %d entries remain" % (size, n)
    if len(signatures) == 0:
        if n == 0:
            return "unsigned", "the ledger is empty and unsigned: nothing is attested"
        return "unsigned", "%d entries chain consistently but are UNSIGNED: consistency is not attribution" % n
    by_seq = {}
    for s in signatures:
        by_seq[s["seq"]] = s
    for e in entries:
        s = by_seq.get(e["seq"])
        if s is None:
            return "signature-gap", "entry %d has no signature: %d of %d entries are signed" % (
                e["seq"],
                len(signatures),
                n,
            )
        if s["entryHash"] != e["entryHash"]:
            return "edited", "entry %d does not match the hash its signature covers" % e["seq"]
        if s["keyId"] != pinned_key_id:
            return "untrusted-key", "entry %d is signed by keyId %s, which is not the pinned key" % (
                e["seq"],
                short_id(s["keyId"]),
            )
        raw = strict_b64(s["signature"])
        msg = canonical({"schemaVersion": 1, "seq": s["seq"], "entryHash": s["entryHash"]}).encode("utf-8")
        if raw is None or not ed25519_verify(pinned_raw, msg, raw):
            return "forged", "entry %d carries a signature that does not verify under %s" % (e["seq"], pinned_key_id)
    return "verified", "%d entries chain from genesis and every one is signed by %s" % (n, pinned_key_id)


def check_texts(ledger_text, sig_text, key_text, head_text=None, bundle_text=None):
    """Everything main() decides once the files are read. Returns (status, detail, key_id, entries).

    status is a ladder verdict, or "unreadable" (exit 2) when the key is not an Ed25519 public key."""
    try:
        raw, kid = load_key(key_text)
    except ValueError as exc:
        return "unreadable", "the public key cannot be read (%s)" % exc, None, 0
    try:
        entries = parse_ledger(ledger_text)
    except Malformed as exc:
        return "malformed-ledger", str(exc), kid, 0
    status, detail = verify_ledger(
        entries,
        parse_signatures(sig_text),
        raw,
        kid,
        read_optional(head_text),
        read_optional(bundle_text),
    )
    return status, detail, kid, len(entries)


def _read(path):
    # newline="" keeps the bytes as written: no line-ending translation, exactly as a browser reads them.
    with open(path, "r", encoding="utf-8", newline="") as fh:
        return fh.read()


def _printable(text):
    """Input can reach a message (a key id, say); never let it drive the terminal."""
    return "".join(c if " " <= c <= "~" else "\\u%04x" % ord(c) if ord(c) < 0x10000 else "\\U%08x" % ord(c) for c in text)


def main(argv):
    args, opts = [], {}
    i = 0
    while i < len(argv):
        if argv[i] in ("--tree-head", "--bundle") and i + 1 < len(argv):
            opts[argv[i]] = argv[i + 1]
            i += 2
        elif argv[i].startswith("--"):
            print(__doc__)
            return 2
        else:
            args.append(argv[i])
            i += 1
    if len(args) != 3:
        print(__doc__)
        return 2
    try:
        ledger_text, sig_text, key_text = _read(args[0]), _read(args[1]), _read(args[2])
        head_text = _read(opts["--tree-head"]) if "--tree-head" in opts else None
        bundle_text = _read(opts["--bundle"]) if "--bundle" in opts else None
    except (OSError, ValueError) as exc:
        print("could not read input: %s" % _printable(str(exc)))
        return 2
    status, detail, kid, count = check_texts(ledger_text, sig_text, key_text, head_text, bundle_text)
    if status == "unreadable":
        print("could not read input: %s" % _printable(detail))
        return 2
    print(_printable("%s  %-20s %s" % ("PASS" if status == "verified" else "FAIL", status, detail)))
    if status == "verified":
        print("VERIFIED - %d entries, signed by %s" % (count, kid))
    elif status == "unsigned":
        print("UNSIGNED - consistent, but not attributed to any key")
    else:
        print("NOT VERIFIED - %s" % status)
    return EXIT.get(status, 4)


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))