paperspec/tools/oracle/receipts.py

"""Receipts, revocations, levels, spot-check sampling and reviewer error bounds.

Non-normative; see spec/08-fidelity.md.
"""

import base64
import hashlib
import math
from fractions import Fraction

from . import crypto, jcs
from .model import LEVELS
from .text import record_id, sha256_hex

SIGNED_FORMATS = ("scienting-receipt/1", "scienting-revocation/1", "scienting-policy/1")


def record_hash(rec):
    """SHA-256 of the canonical JSON of a signed record without its `id` and `signature`."""
    return sha256_hex(jcs.dumpb({k: v for k, v in rec.items() if k not in ("id", "signature")}))


def signing_payload(rec):
    return jcs.dumpb({"format": rec["format"], "hash": record_hash(rec)})


def seal(rec, secret, key_id):
    """Fill in a record's root (for receipts), ID and Ed25519 signature."""
    rec = dict(rec)
    if rec["format"] == "scienting-receipt/1":
        rec["root"] = crypto.merkle_root(crypto.receipt_items(rec))
    rec.pop("id", None)
    rec.pop("signature", None)
    sig = crypto.ed25519_sign(secret, signing_payload(rec))
    rec["id"] = record_id(record_hash(rec))
    rec["signature"] = {"scheme": "ed25519", "key": key_id, "value": base64.b64encode(sig).decode("ascii")}
    return rec


def verify_signature(rec, roster):
    sig = rec.get("signature") or {}
    key = roster.get(sig.get("key"))
    if not key or sig.get("scheme") != "ed25519" or key["scheme"] != "ed25519":
        return False
    pub = base64.b64decode(key["public"])
    return crypto.ed25519_verify(pub, signing_payload(rec), base64.b64decode(sig["value"]))


def key_valid_at(key, at):
    return key["registered_at"] <= at and (key.get("revoked_at") is None or at < key["revoked_at"])


# --- spot-checks --------------------------------------------------------------


def seed_payload(receipt):
    return jcs.dumpb({"format": "scienting-spot-check/1", "receipt": record_hash(receipt)})


def make_seed(receipt, checker_secret):
    return crypto.ed25519_sign(checker_secret, seed_payload(receipt))


def sample_size(n, rate="0.05", minimum=10):
    return min(n, max(minimum, math.ceil(Fraction(rate) * n)))


def sample(receipt, seed, rate="0.05", minimum=10):
    ids = [nid for nid, _ in receipt["nodes"]]
    m = sample_size(len(ids), rate, minimum)
    ranked = sorted(ids, key=lambda nid: (hashlib.sha256(seed + nid.encode("utf-8")).hexdigest(), nid))
    return sorted(ranked[:m])


def upper_bound(k, n, confidence="0.95"):
    """One-sided Clopper-Pearson upper bound on an error rate, as a float."""
    if n <= 0:
        raise ValueError("n must be positive")
    if k >= n:
        return 1.0
    alpha = 1.0 - float(Fraction(confidence))

    def cdf(p):
        if p <= 0:
            return 1.0
        if p >= 1:
            return 0.0
        lp, lq = math.log(p), math.log1p(-p)
        total = 0.0
        for i in range(k + 1):
            total += math.exp(math.lgamma(n + 1) - math.lgamma(i + 1) - math.lgamma(n - i + 1) + i * lp
                              + (n - i) * lq)
        return total

    lo, hi = 0.0, 1.0
    for _ in range(200):
        mid = (lo + hi) / 2
        if cdf(mid) > alpha:
            lo = mid
        else:
            hi = mid
    return hi


def bound_string(k, n, confidence="0.95"):
    """The bound as the spec reports it: six decimal places, rounded up."""
    x = upper_bound(k, n, confidence)
    v = math.ceil(x * 10**6 - 1e-9)
    return "%d.%06d" % (v // 10**6, v % 10**6)


# --- levels -------------------------------------------------------------------


def revoked(receipt, revocations):
    for r in revocations:
        if r.get("receipt") == receipt["id"]:
            return True
        if r.get("signer") == receipt["by"] and r.get("period"):
            p = r["period"]
            if p["from"] <= receipt["at"] <= p["to"]:
                return True
    return False


def valid_revocations(revocations, receipts_by_id, roster):
    """The revocations signed by the receipt's signer, the signer they name, or a key the roster makes an owner."""
    owners = {k for k, v in roster.items() if v.get("role") == "owner"}
    out = []
    for r in revocations:
        if not verify_signature(r, roster):
            continue
        signer = r["signature"]["key"]
        allowed = set(owners)
        if r.get("receipt") in receipts_by_id:
            allowed.add(receipts_by_id[r["receipt"]]["by"])
        if r.get("signer"):
            allowed.add(r["signer"])
        if signer in allowed:
            out.append(r)
    return out


def receipt_status(rec, records, roster):
    """`valid`, `revoked`, `suspended`, or `invalid` for one receipt."""
    receipts = [x for x in records if x["format"] == "scienting-receipt/1"]
    revs = [x for x in records if x["format"] == "scienting-revocation/1"]
    by_id = {x["id"]: x for x in receipts}
    if rec.get("root") != crypto.merkle_root(crypto.receipt_items(rec)):
        return "invalid"
    if rec["id"] != record_id(record_hash(rec)) or not verify_signature(rec, roster):
        return "invalid"
    key = roster.get(rec["by"])
    if not key or not key_valid_at(key, rec["at"]) or rec["signature"]["key"] != rec["by"]:
        return "invalid"
    if revoked(rec, valid_revocations(revs, by_id, roster)):
        return "revoked"
    for q in receipts:
        qa = q.get("qa")
        if qa and qa.get("of") == rec["id"] and qa.get("errors"):
            if receipt_status(q, records, roster) == "valid":
                return "suspended"
    return "valid"


def effective_levels(nodes, values, records, roster):
    """Every leaf's level: its recorded level, raised by valid receipts that list it at its current hash.

    A spot-check's `qa:` receipt does not raise the nodes it found in error (spec/08 §8.2).
    """
    level = {n["id"]: n.get("fidelity") for n in nodes if "fidelity" in n}
    for rec in records:
        if rec["format"] != "scienting-receipt/1":
            continue
        if receipt_status(rec, records, roster) != "valid":
            continue
        wrong = {e["node"] for e in (rec.get("qa") or {}).get("errors", [])}
        for nid, h in rec.get("nodes", []):
            if nid in level and nid not in wrong and values[nid]["node_hash"] == h:
                if LEVELS.index(rec["level"]) > LEVELS.index(level[nid]):
                    level[nid] = rec["level"]
    return level


def distribution(levels):
    counts = {lv: 0 for lv in LEVELS}
    for lv in levels.values():
        counts[lv] += 1
    lowest = next((lv for lv in LEVELS if counts[lv]), None)
    return {"lowest": lowest, "counts": counts}