paperspec/tools/oracle/crypto.py

"""Keyed hashing, key derivation, Merkle roots and Ed25519.

Non-normative; see spec/05-hashes.md and spec/08-fidelity.md.
HKDF and Ed25519 come from the `cryptography` package, imported on first use so that the hashing, Merkle
and did:key functions work without it.
"""

import hashlib

from . import jcs


def _cryptography():
    """The `cryptography` modules HKDF and Ed25519 use: exceptions, hashes, ed25519 and hkdf."""
    try:
        from cryptography import exceptions
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import ed25519
        from cryptography.hazmat.primitives.kdf import hkdf
    except ImportError:
        raise ImportError("HKDF and Ed25519 need the cryptography package: install paperspec[crypto]") from None
    return exceptions, hashes, ed25519, hkdf


def hkdf(ikm, salt, info, length=32):
    """HKDF-SHA-256, RFC 5869."""
    _, hashes, _, kdf = _cryptography()
    return kdf.HKDF(algorithm=hashes.SHA256(), length=length, salt=salt, info=info).derive(ikm)


def team_key(secret):
    return hkdf(secret, b"scienting-team-key/1", b"scienting-hash/1")


def key_id(key):
    return hashlib.sha256(b"scienting-key-id/1" + key).hexdigest()[:16]


def page_hash(page_fp_hex):
    return hashlib.sha256(b"scienting-page-hash/1" + bytes.fromhex(page_fp_hex)).hexdigest()


def page_key(page_fps_hex, context):
    ikm = b"".join(bytes.fromhex(h) for h in page_fps_hex)
    return hkdf(ikm, b"scienting-page-key/1", context.encode("utf-8"))


# --- Merkle trees (RFC 6962 hashing) ------------------------------------------


def _leaf(data):
    return hashlib.sha256(b"\x00" + data).digest()


def _node(left, right):
    return hashlib.sha256(b"\x01" + left + right).digest()


def _mth(leaves):
    n = len(leaves)
    if n == 1:
        return leaves[0]
    k = 1
    while k * 2 < n:
        k *= 2
    return _node(_mth(leaves[:k]), _mth(leaves[k:]))


def merkle_root(items):
    """Root over receipt items: each item is a list such as ["node", id, hash]."""
    if not items:
        raise ValueError("a Merkle root needs at least one item")
    return _mth([_leaf(jcs.dumpb(list(it))) for it in items]).hex()


def receipt_items(receipt):
    items = [["node", nid, h] for nid, h in receipt.get("nodes", [])]
    items += [list(r) for r in receipt.get("records", [])]
    return items


EMPTY_ROOT = hashlib.sha256(b"").hexdigest()


def history_root(items):
    """Root over an append-only history, with RFC 9162's hash of the empty list for no items."""
    if not items:
        return EMPTY_ROOT
    return _mth([_leaf(jcs.dumpb(list(it))) for it in items]).hex()


def leaf_hash(item):
    return _leaf(jcs.dumpb(list(item))).hex()


# --- RFC 9162 proofs (§2.1.3 and §2.1.4) ---------------------------------------


def _path(m, leaves):
    """PATH(m, D[n]): the inclusion path for the leaf at index m, as RFC 9162 §2.1.3.1 defines it."""
    n = len(leaves)
    if n == 1:
        return []
    k = 1
    while k * 2 < n:
        k *= 2
    if m < k:
        return _path(m, leaves[:k]) + [_mth(leaves[k:])]
    return _path(m - k, leaves[k:]) + [_mth(leaves[:k])]


def inclusion_path(items, index):
    """The audit path of `items[index]` in the tree over `items`, as hex hashes, leaf-most first."""
    if not 0 <= index < len(items):
        raise ValueError("index outside the tree")
    return [h.hex() for h in _path(index, [_leaf(jcs.dumpb(list(it))) for it in items])]


def verify_inclusion(root, leaf, index, size, path):
    """Whether `path` proves that the leaf hash `leaf` is item `index` of a tree of `size` with `root`
    (RFC 9162 §2.1.3.2). Never raises."""
    try:
        if not 0 <= index < size:
            return False
        fn, sn = index, size - 1
        r = bytes.fromhex(leaf)
        for p in path:
            p = bytes.fromhex(p)
            if sn == 0:
                return False
            if fn & 1 or fn == sn:
                r = _node(p, r)
                while fn & 1 == 0 and fn != 0:
                    fn >>= 1
                    sn >>= 1
            else:
                r = _node(r, p)
            fn >>= 1
            sn >>= 1
        return sn == 0 and r.hex() == root
    except (ValueError, TypeError):
        return False


def _subproof(m, leaves, b):
    n = len(leaves)
    if m == n:
        return [] if b else [_mth(leaves)]
    k = 1
    while k * 2 < n:
        k *= 2
    if m <= k:
        return _subproof(m, leaves[:k], b) + [_mth(leaves[k:])]
    return _subproof(m - k, leaves[k:], False) + [_mth(leaves[:k])]


def consistency_proof(items, m):
    """PROOF(m, D[n]): what shows the tree over the first m items is a prefix of the tree over all n
    (RFC 9162 §2.1.4.1), as hex hashes; empty when m is 0 or n."""
    n = len(items)
    if not 0 <= m <= n:
        raise ValueError("m outside the tree")
    if m == 0 or m == n:
        return []
    return [h.hex() for h in _subproof(m, [_leaf(jcs.dumpb(list(it))) for it in items], True)]


def verify_consistency(root_m, root_n, m, n, proof):
    """Whether `proof` shows the tree of size m with `root_m` is a prefix of the tree of size n with `root_n`
    (RFC 9162 §2.1.4.2). Never raises."""
    try:
        if m == 0:
            return not proof and root_m == EMPTY_ROOT
        if m == n:
            return not proof and root_m == root_n
        if not 0 < m < n or not proof:
            return False
        proof = [bytes.fromhex(p) for p in proof]
        fn, sn = m - 1, n - 1
        while fn & 1:
            fn >>= 1
            sn >>= 1
        if fn == 0:
            fr = sr = bytes.fromhex(root_m)
        else:
            fr = sr = proof[0]
            proof = proof[1:]
        for c in proof:
            if sn == 0:
                return False
            if fn & 1 or fn == sn:
                fr = _node(c, fr)
                sr = _node(c, sr)
                while fn & 1 == 0 and fn != 0:
                    fn >>= 1
                    sn >>= 1
            else:
                sr = _node(sr, c)
            fn >>= 1
            sn >>= 1
        return sn == 0 and fr.hex() == root_m and sr.hex() == root_n
    except (ValueError, TypeError):
        return False


# --- Ed25519 (RFC 8032) -------------------------------------------------------


def ed25519_public(secret):
    """The 32-byte public key of a 32-byte Ed25519 secret."""
    _, _, ed25519, _ = _cryptography()
    return ed25519.Ed25519PrivateKey.from_private_bytes(secret).public_key().public_bytes_raw()


def ed25519_sign(secret, msg):
    _, _, ed25519, _ = _cryptography()
    return ed25519.Ed25519PrivateKey.from_private_bytes(secret).sign(msg)


def ed25519_verify(public, msg, signature):
    """True when `signature` is a valid Ed25519 signature of `msg` under `public`; never raises."""
    if len(public) != 32 or len(signature) != 64:
        return False
    exceptions, _, ed25519, _ = _cryptography()
    try:
        ed25519.Ed25519PublicKey.from_public_bytes(public).verify(signature, msg)
    except (exceptions.InvalidSignature, ValueError):
        return False
    return True


# --- did:key (W3C did:key method, multicodec and base58btc) ---------------------------

BASE58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
MULTICODEC = {"ed25519": 0xED, "ml-dsa-65": 0x1207}


def _varint(n):
    out = bytearray()
    while True:
        b = n & 0x7F
        n >>= 7
        out.append(b | (0x80 if n else 0))
        if not n:
            return bytes(out)


def base58btc(data):
    n = int.from_bytes(data, "big")
    out = ""
    while n:
        n, r = divmod(n, 58)
        out = BASE58[r] + out
    return "1" * (len(data) - len(data.lstrip(b"\x00"))) + out


def did_key(scheme, public):
    """The did:key of a public key: `z` and the base58btc of the multicodec prefix and the key bytes."""
    if scheme not in MULTICODEC:
        raise ValueError("no multicodec code for " + scheme)
    return "did:key:z" + base58btc(_varint(MULTICODEC[scheme]) + public)