#!/usr/bin/env python3
"""Check a vCISO Lite seal receipt without trusting vCISO Lite.

A seal receipt (receipt.json, downloaded from a seal's page) carries what's
needed to repeat the seal's checks offline:

  1. the signature over the record's fingerprint (Ed25519, RFC 8032);
  2. the record's path to its checkpoint's Merkle root;
  3. the checkpoint's own hash, recomputed from its published fields;
  4. the checkpoint's path to a public log head (RFC 6962);
  5. the outside timestamps on the checkpoint and the head (RFC 3161).

This script runs checks 1 to 4 and reads the timestamps. It can't check the
timestamp authority's signature with the standard library alone, so it writes
each timestamp to a file and prints the openssl command that does.

Python 3.8+, standard library only. Read it before you run it.

  python3 verify_receipt.py receipt.json
  python3 verify_receipt.py receipt.json sealed-record-41.json
  python3 verify_receipt.py receipt.json --offline --keys keys.json

The optional second file is a sealed record you downloaded. Its SHA-256 must
equal the receipt's fingerprint. Without --offline, the script fetches the
signing keys and the log head the receipt names, and compares them.
"""
import base64
import hashlib
import json
import sys
import urllib.error
import urllib.request

FAILED = []


def report(ok, what, detail=""):
    print(("PASS  " if ok else "FAIL  ") + what + (": " + detail if detail else ""))
    if not ok:
        FAILED.append(what)


def note(what):
    print("NOTE  " + what)


def fetch_json(url):
    req = urllib.request.Request(url, headers={"User-Agent": "verify_receipt.py"})
    with urllib.request.urlopen(req, timeout=20) as resp:
        return json.load(resp)


# --- Ed25519 verification, RFC 8032 section 5.1.7 ---------------------------

P = 2 ** 255 - 19
L = 2 ** 252 + 27742317777372353535851937790883648493
D = -121665 * pow(121666, P - 2, P) % P
SQRT_M1 = pow(2, (P - 1) // 4, P)
BASE = (
    15112221349535400772501151409588531511454012693041857206046113283949847762202,
    46316835694926478169428394003475163141307993866256225615783033603165251855960,
)


def _ext(pt):
    x, y = pt
    return (x, y, 1, x * y % P)


def _add(a, b):
    x1, y1, z1, t1 = a
    x2, y2, z2, t2 = b
    pa = (y1 - x1) * (y2 - x2) % P
    pb = (y1 + x1) * (y2 + x2) % P
    pc = t1 * 2 * D * t2 % P
    pd = z1 * 2 * z2 % P
    e, f, g, h = pb - pa, pd - pc, pd + pc, pb + pa
    return (e * f % P, g * h % P, f * g % P, e * h % P)


def _mul(s, pt):
    acc = (0, 1, 1, 0)
    while s:
        if s & 1:
            acc = _add(acc, pt)
        pt = _add(pt, pt)
        s >>= 1
    return acc


def _equal(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 _decompress(raw):
    if len(raw) != 32:
        return None
    y = int.from_bytes(raw, "little")
    sign = y >> 255
    y &= (1 << 255) - 1
    if y >= P:
        return None
    x2 = (y * y - 1) * pow(D * y * y + 1, P - 2, P) % P
    if x2 == 0:
        return None if sign else (0, y, 1, 0)
    x = pow(x2, (P + 3) // 8, P)
    if (x * x - x2) % P:
        x = x * SQRT_M1 % P
    if (x * x - x2) % P:
        return None
    if x & 1 != sign:
        x = P - x
    return (x, y, 1, x * y % P)


def ed25519_verify(public_key, message, signature):
    if len(public_key) != 32 or len(signature) != 64:
        return False
    a = _decompress(public_key)
    r = _decompress(signature[:32])
    s = int.from_bytes(signature[32:], "little")
    if a is None or r is None or s >= L:
        return False
    h = int.from_bytes(hashlib.sha512(signature[:32] + public_key + message).digest(), "little") % L
    return _equal(_mul(s, _ext(BASE)), _add(r, _mul(h, a)))


# --- Merkle paths -----------------------------------------------------------

def fold(tree_epoch, leaf_hex, path):
    """Fold an inclusion path from a leaf to a root, returning the root as hex.

    Epoch 2 is RFC 6962: raw 32-byte values, 0x00 before a leaf, 0x01 before
    a node. Epoch 1 (older checkpoints) hashes the two children's hex strings
    concatenated as text.
    """
    if tree_epoch == 2:
        running = hashlib.sha256(b"\x00" + bytes.fromhex(leaf_hex)).digest()
        for step in path:
            sib = bytes.fromhex(step["sibling_hash"])
            pair = running + sib if step["right"] else sib + running
            running = hashlib.sha256(b"\x01" + pair).digest()
        return running.hex()
    if tree_epoch == 1:
        running = leaf_hex
        for step in path:
            sib = step["sibling_hash"]
            pair = running + sib if step["right"] else sib + running
            running = hashlib.sha256(pair.encode()).hexdigest()
        return running
    raise ValueError("unknown tree_epoch %r" % tree_epoch)


# --- Just enough DER to read an RFC 3161 token ------------------------------

TSTINFO_OID = bytes.fromhex("060b2a864886f70d0109100104")  # id-ct-TSTInfo


def _tlv(buf, i):
    tag = buf[i]
    n = buf[i + 1]
    i += 2
    if n & 0x80:
        count = n & 0x7F
        n = int.from_bytes(buf[i:i + count], "big")
        i += count
    return tag, buf[i:i + n], i + n


def read_timestamp(tsr):
    """Return (imprint_hex, gen_time) from a TimeStampResp, or raise."""
    at = tsr.find(TSTINFO_OID)
    if at < 0:
        raise ValueError("no TSTInfo in the token")
    _, explicit, _ = _tlv(tsr, at + len(TSTINFO_OID))     # [0] EXPLICIT
    _, tstinfo_der, _ = _tlv(explicit, 0)                 # OCTET STRING
    _, tstinfo, _ = _tlv(tstinfo_der, 0)                  # TSTInfo SEQUENCE
    i = 0
    _, _, i = _tlv(tstinfo, i)                            # version
    _, _, i = _tlv(tstinfo, i)                            # policy
    _, imprint, i = _tlv(tstinfo, i)                      # messageImprint
    _, _, j = _tlv(imprint, 0)                            #   hashAlgorithm
    _, hashed, _ = _tlv(imprint, j)                       #   hashedMessage
    _, _, i = _tlv(tstinfo, i)                            # serialNumber
    tag, gen_time, _ = _tlv(tstinfo, i)                   # genTime
    if tag != 0x18:
        raise ValueError("genTime not where expected")
    return hashed.hex(), gen_time.decode()


def check_timestamp(label, b64, root_hex, filename):
    try:
        raw = base64.b64decode(b64)
        imprint, when = read_timestamp(raw)
    except (ValueError, IndexError) as exc:
        report(False, label + " timestamp is readable", str(exc))
        return
    report(imprint == root_hex, label + " timestamp covers its root", "stamped " + when)
    with open(filename, "wb") as fh:
        fh.write(raw)
    print("      to check the authority's signature too:")
    print("      openssl ts -verify -in %s -digest %s -CAfile <your CA bundle>" % (filename, root_hex))


# --- The checks -------------------------------------------------------------

def main(argv):
    args = [a for a in argv[1:] if not a.startswith("--")]
    offline = "--offline" in argv
    keys_file = None
    if "--keys" in argv:
        keys_file = argv[argv.index("--keys") + 1]
        args.remove(keys_file)
    if not args:
        print(__doc__)
        return 2

    rc = json.load(open(args[0]))
    if rc.get("format") != "vcisolite-seal-receipt/1":
        print("Refused: not a vCISO Lite seal receipt (format %r)" % rc.get("format"))
        return 2

    state = rc["state"]
    fingerprint = rc["fingerprint_sha256"]
    seal = rc["seal"]
    print("seal #%s, sealed %s, state %s" % (seal["sequence_number"], seal["sealed_at"], state))
    print("fingerprint " + fingerprint)
    print()

    # Your copy of the record.
    if len(args) > 1:
        mine = hashlib.sha256(open(args[1], "rb").read()).hexdigest()
        report(mine == fingerprint, "your sealed record matches the fingerprint", mine)
    if state == "legacy":
        note("this record was sealed before content fingerprints; its contents can't be checked")

    # 1. The signature.
    if rc.get("signature") and rc.get("key_id"):
        try:
            keys = json.load(open(keys_file)) if keys_file else (None if offline else fetch_json(rc["keys_url"]))
        except (OSError, urllib.error.URLError, ValueError) as exc:
            keys = None
            report(False, "signing keys fetched", str(exc))
        if keys is not None:
            jwk = next((k for k in keys.get("keys", []) if k.get("kid") == rc["key_id"]), None)
            if jwk is None:
                report(False, "signing key %s is published" % rc["key_id"])
            else:
                pub = base64.urlsafe_b64decode(jwk["x"] + "=" * (-len(jwk["x"]) % 4))
                msg = ("vciso-audit-entry-v1:%s:%s" % (rc["chain"], fingerprint)).encode()
                sig = base64.b64decode(rc["signature"])
                report(ed25519_verify(pub, msg, sig), "signature by key " + rc["key_id"])
        elif offline:
            note("signature not checked: pass --keys keys.json when offline")
    else:
        note("this record is unsigned; nothing to check for the signature")

    if state == "awaiting_checkpoint":
        note("not in a checkpoint yet; download a new receipt after the next checkpoint")
        return finish()

    # 2. Record -> checkpoint root.
    cp = rc["checkpoint"]
    rp = rc["record_proof"]
    root = fold(cp["tree_epoch"], fingerprint, rp["audit_path"])
    report(root == cp["merkle_root"], "record is in checkpoint #%s" % cp["checkpoint_number"],
           "leaf %s of records #%s-#%s" % (rp["leaf_index"], cp["first_sequence_number"], cp["last_sequence_number"]))

    # 3. The checkpoint's own hash, from its published fields.
    canonical = base64.b64decode(cp["canonical_b64"])
    report(hashlib.sha256(canonical).hexdigest() == cp["checkpoint_hash"], "checkpoint hash recomputes")
    fields = json.loads(canonical)
    report(fields.get("merkle_root") == cp["merkle_root"], "checkpoint hash covers that root")
    if "chain_id" in fields:
        report(fields["chain_id"] == rc["chain"], "checkpoint is on this record's chain")
    else:
        note("this checkpoint predates chain ids in checkpoint hashes; its chain is bound by the record proof")
    report(fields.get("previous_checkpoint_hash") == rc["previous_checkpoint"],
           "checkpoint links to the one before it", (rc["previous_checkpoint"] or "")[:16] + "...")
    if rc.get("checkpoint_tsr"):
        check_timestamp("checkpoint", rc["checkpoint_tsr"], cp["merkle_root"], "checkpoint.tsr")
    else:
        note("the checkpoint has no timestamp of its own; the log head's timestamp below bounds it")

    if state == "awaiting_log":
        note("checkpoint not in the public log yet; download a new receipt later")
        return finish()

    # 4. Checkpoint -> public log head.
    lp = rc["log_proof"]
    log = rc["log"]
    report(lp["leaf_hash"] == cp["checkpoint_hash"], "log entry is this checkpoint")
    head_root = fold(lp["tree_epoch"], lp["leaf_hash"], lp["audit_path"])
    report(head_root == log["root_hash"], "checkpoint is entry %s in the public log" % lp["tree_index"],
           "under head %s" % log["tree_size"])
    check_timestamp("log head", log["head_tsr"], log["root_hash"], "head.tsr")
    if offline:
        note("head not compared with the published log (--offline): fetch " + log["head_url"])
    else:
        try:
            pub = fetch_json(log["head_url"])
            same = pub.get("root_hash") == log["root_hash"] and pub.get("tree_size") == log["tree_size"]
            report(same, "head matches the published log", log["head_url"])
        except (urllib.error.URLError, ValueError) as exc:
            report(False, "published head fetched", str(exc))
    print()
    print("Keep head %s. Later, verify_consistency.py %s proves the log was only appended to."
          % (log["tree_size"], log["tree_size"]))
    return finish()


def finish():
    print()
    if FAILED:
        print("FAILED: " + "; ".join(FAILED))
        return 1
    print("All checks passed.")
    return 0


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