"""A team's transparency log of receipts, revocations and roster changes, and its signed checkpoints. Non-normative; see spec/23-log.md. Entries are the items of an RFC 9162 tree (the same tree as receipts and the correction history), and a checkpoint is a signed note in the format transparency logs share (C2SP `signed-note` and `tlog-checkpoint`): an origin line, the tree size, the root hash in Base64, an empty line, and one signature line per signer. """ import base64 import hashlib from . import crypto, jcs ALGORITHM_ED25519 = 0x01 FORMAT = "scienting-checkpoint/1" def entry(record): """The log item for a signed record: its kind, its ID and its record hash.""" kinds = {"scienting-receipt/1": "receipt", "scienting-revocation/1": "revocation", "scienting-roster/1": "roster"} kind = kinds.get(record.get("format")) if kind is None: raise ValueError("only receipts, revocations and rosters are logged") if kind == "roster": return ["roster", record.get("store", ""), hashlib.sha256(jcs.dumpb(record)).hexdigest()] body = {k: v for k, v in record.items() if k not in ("id", "signature")} return [kind, record["id"], hashlib.sha256(jcs.dumpb(body)).hexdigest()] def root(entries): return crypto.history_root(entries) def key_hash(name, public): """The four-byte key ID of a signed-note signer: SHA-256 of the name, a newline, the algorithm byte and the public key, truncated.""" return hashlib.sha256(name.encode("utf-8") + b"\n" + bytes([ALGORITHM_ED25519]) + public).digest()[:4] def _b64(text): """Canonical standard Base64 only: the bytes, or None when the text is not exactly their encoding.""" try: raw = base64.b64decode(text, validate=True) except (ValueError, TypeError): return None return raw if base64.b64encode(raw).decode("ascii") == text else None def note_body(origin, size, root_hex): return "%s\n%d\n%s\n" % (origin, size, base64.b64encode(bytes.fromhex(root_hex)).decode("ascii")) def sign_checkpoint(origin, entries, signers): """A checkpoint note over `entries`, signed by each (name, secret) in `signers`.""" body = note_body(origin, len(entries), root(entries)) lines = [] for name, secret in signers: public = crypto.ed25519_public(secret) sig = crypto.ed25519_sign(secret, body.encode("utf-8")) lines.append("— %s %s" % (name, base64.b64encode(key_hash(name, public) + sig).decode("ascii"))) return body + "\n" + "".join(ln + "\n" for ln in lines) def parse_checkpoint(text): """The origin, size, root (hex) and [(name, key hash, signature)] of a note, or None when malformed.""" try: body, _, sigs = text.partition("\n\n") origin, size, root_b64 = body.split("\n") rt = _b64(root_b64) if rt is None or len(rt) != 32 or not size.isdigit() or str(int(size)) != size or not origin: return None out = [] for ln in sigs.splitlines(): if not ln.startswith("— "): return None _, name, blob = ln.split(" ", 2) raw = _b64(blob) if raw is None or len(raw) != 68: return None out.append((name, raw[:4], raw[4:])) return {"origin": origin, "size": int(size), "root": rt.hex(), "signatures": out} except (ValueError, TypeError): return None def verify_checkpoint(text, roster): """The signers of the note whose Ed25519 signatures verify under the roster's keys of those names. Returns (parsed note, [names verified]); the note is None when malformed. Never raises. """ note = parse_checkpoint(text) if note is None: return None, [] body = note_body(note["origin"], note["size"], note["root"]).encode("utf-8") verified = [] for name, kh, sig in note["signatures"]: key = roster.get(name) if not key or key.get("scheme") != "ed25519": continue try: public = base64.b64decode(key["public"], validate=True) except (ValueError, TypeError): continue if key_hash(name, public) == kh and crypto.ed25519_verify(public, body, sig): verified.append(name) return note, verified