#!/usr/bin/env python3 """Independent verifier of the QoE aggregation corpus (extensions/telemetry-aggregation.md). Written from the extension's text and from the texts it cites: draft-ietf-ppm-dap-20 (DAP), draft-irtf-cfrg-vdaf-22 (the VDAF draft, whose algorithms are written in Python and are followed here), RFC 9180 (HPKE), RFC 9861 (TurboSHAKE), RFC 7748 (X25519), RFC 5869 (HKDF), FIPS 197 (AES) and NIST SP 800-38D (GCM). It shares no code with wai-rs, which generates the corpus, and never runs it. Ed25519 signatures are checked with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*). Everything else is here: - Keccak-p[1600, 12], TurboSHAKE128 and the VDAF draft's XofTurboShake128 (§6.2.1); - Field64 and Field128, the Lagrange-basis polynomials, the gadgets and circuits of Prio3Count, Prio3Sum, Prio3SumVec and Prio3Histogram, the FLP and Prio3 itself (§6, §7, Appendix A), checked first against the draft's own test vectors in vdaf-22/; - X25519, HKDF-SHA256, AES-128-GCM and HPKE base mode, checked against RFC 9180 A.1.1; - DAP's messages, and each flow of the corpus re-run from its reports: every input share opened with the aggregators' test keys, each measurement recombined and required to be its session's (extension §3.1), both aggregators' verification, the aggregation job's request and response required byte for byte, the batch buckets, every refusal, the collection, every aggregate share and statement opened with the collector's key, and the result; - the statement and the aggregate receipt (§6, §7), verification in §8.1's order with each refusal's place, and a room's cell (§9.2); - delivery-telemetry §3.4's key classes, read from that text, and §4.3's delivery-only reports. It requires the outcome the corpus states for everything, that every reason of §11 (read from the extension) and every code it reuses appears, then mutates copies of the corpus, one defect each, and requires every copy to fail; and it drops each of its own checks in turn and requires the corpus to catch the weakened verifier. python3 telemetry_aggregate_verify.py Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier fails. Python 3.8 or later; standard library only. """ import copy import hashlib import hmac import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance")) import _ed25519 as ED # noqa: E402 EXTENSION = os.path.join(HERE, "..", "extensions", "telemetry-aggregation.md") TELEMETRY = os.path.join(HERE, "..", "extensions", "delivery-telemetry.md") SKIP = set() class Refuse(Exception): def __init__(self, reason, at=None): super().__init__(reason) self.reason, self.at = reason, at def sha256(b): return hashlib.sha256(b).digest() # --------------------------------------------------------------------------------------- # Keccak-p[1600, 12] and TurboSHAKE128 (RFC 9861 §2). ROUND_CONSTANTS = [ 0x0000000000000001, 0x0000000000008082, 0x800000000000808A, 0x8000000080008000, 0x000000000000808B, 0x0000000080000001, 0x8000000080008081, 0x8000000000008009, 0x000000000000008A, 0x0000000000000088, 0x0000000080008009, 0x000000008000000A, 0x000000008000808B, 0x800000000000008B, 0x8000000000008089, 0x8000000000008003, 0x8000000000008002, 0x8000000000000080, 0x000000000000800A, 0x800000008000000A, 0x8000000080008081, 0x8000000000008080, 0x0000000080000001, 0x8000000080008008, ] ROTATIONS = [[0, 36, 3, 41, 18], [1, 44, 10, 45, 2], [62, 6, 43, 15, 61], [28, 55, 25, 21, 56], [27, 20, 39, 8, 14]] M64 = (1 << 64) - 1 PI = [(y + 5 * ((2 * x + 3 * y) % 5), x + 5 * y, ROTATIONS[x][y]) for x in range(5) for y in range(5)] def keccak_p12(a): for rc in ROUND_CONSTANTS[12:]: c = [a[x] ^ a[x + 5] ^ a[x + 10] ^ a[x + 15] ^ a[x + 20] for x in range(5)] d = [c[(x - 1) % 5] ^ (((c[(x + 1) % 5] << 1) | (c[(x + 1) % 5] >> 63)) & M64) for x in range(5)] a = [a[i] ^ d[i % 5] for i in range(25)] b = [0] * 25 for dst, src, r in PI: v = a[src] b[dst] = ((v << r) | (v >> (64 - r))) & M64 if r else v a = [b[i] ^ ((~b[(i + 1) % 5 + 5 * (i // 5)]) & b[(i + 2) % 5 + 5 * (i // 5)]) for i in range(25)] a[0] ^= rc return a class TurboShake128: RATE = 168 def __init__(self, m, d): a = [0] * 25 p = bytearray(m) + bytes([d]) p += bytes((-len(p)) % self.RATE) p[-1] ^= 0x80 for off in range(0, len(p), self.RATE): blk = p[off:off + self.RATE] for i in range(21): a[i] ^= int.from_bytes(blk[8 * i:8 * i + 8], "little") a = keccak_p12(a) self.a, self.buf = a, b"" def read(self, n): while len(self.buf) < n: self.buf += b"".join(x.to_bytes(8, "little") for x in self.a[:21]) self.a = keccak_p12(self.a) out, self.buf = self.buf[:n], self.buf[n:] return out def turboshake128(m, d, n): return TurboShake128(m, d).read(n) VERSION = 18 # the VDAF draft's §2 def format_dst(algo_class, algo, usage): return bytes([VERSION, algo_class]) + algo.to_bytes(4, "big") + usage.to_bytes(2, "big") class Xof: """XofTurboShake128 (VDAF §6.2.1).""" SEED_SIZE = 32 def __init__(self, seed, dst, binder): self.t = TurboShake128(len(dst).to_bytes(2, "little") + dst + bytes([len(seed)]) + seed + binder, 1) def next(self, n): return self.t.read(n) def next_vec(self, field, length): mask = (1 << (8 * field.SIZE)) - 1 # next_power_of_2(MODULUS) - 1, every bit of the encoding out = [] while len(out) < length: x = int.from_bytes(self.next(field.SIZE), "little") & mask if x < field.P: out.append(x) return out @classmethod def derive_seed(cls, seed, dst, binder): return cls(seed, dst, binder).next(cls.SEED_SIZE) @classmethod def expand_into_vec(cls, field, seed, dst, binder, length): return cls(seed, dst, binder).next_vec(field, length) # --------------------------------------------------------------------------------------- # Fields (VDAF §6.1); elements are plain integers below P. class Field64: P = 2 ** 32 * 4294967295 + 1 SIZE = 8 GEN_ORDER = 2 ** 32 GEN = pow(7, 4294967295, P) class Field128: P = 2 ** 66 * 4611686018427387897 + 1 SIZE = 16 GEN_ORDER = 2 ** 66 GEN = pow(7, 4611686018427387897, P) def inv(f, x): return pow(x, f.P - 2, f.P) def nth_root(f, n): return pow(f.GEN, f.GEN_ORDER // n, f.P) def nth_root_powers(f, n): w = nth_root(f, n) out, x = [], 1 for _ in range(n): out.append(x) x = x * w % f.P return out def encode_vec(f, v): return b"".join(x.to_bytes(f.SIZE, "little") for x in v) def decode_vec(f, b): if len(b) % f.SIZE: raise ValueError("length") v = [int.from_bytes(b[i:i + f.SIZE], "little") for i in range(0, len(b), f.SIZE)] if any(x >= f.P for x in v): raise ValueError("modulus overflow") return v def vadd(f, a, b): if len(a) != len(b): raise ValueError("mismatched vector sizes") return [(x + y) % f.P for x, y in zip(a, b)] def vsub(f, a, b): if len(a) != len(b): raise ValueError("mismatched vector sizes") return [(x - y) % f.P for x, y in zip(a, b)] def next_power_of_2(x): n = 1 while n < x: n *= 2 return n def poly_eval_mono(f, p, x): acc = 0 for c in reversed(p): acc = (acc * x + c) % f.P return acc def ntt(f, p, n, set_s=False): s = nth_root(f, 2 * n) if set_s else 1 return [poly_eval_mono(f, p, s * w % f.P) for w in nth_root_powers(f, n)] def inv_ntt(f, v, n): wi = inv(f, nth_root(f, n)) ninv = inv(f, n) out = [] for j in range(n): step = pow(wi, j, f.P) acc, x = 0, 1 for vi in v: acc = (acc + vi * x) % f.P x = x * step % f.P out.append(acc * ninv % f.P) return out def double_evaluations(f, p): n = len(p) odd = ntt(f, inv_ntt(f, p, n), n, True) return [x for pair in zip(p, odd) for x in pair] def poly_mul(f, p, q): a, b = double_evaluations(f, p), double_evaluations(f, q) return [x * y % f.P for x, y in zip(a, b)] def poly_eval_batched(f, polys, x): """VDAF §6.1.3.2, as the draft writes it.""" n = len(polys[0]) nodes = nth_root_powers(f, n) k = 1 u = [p[0] for p in polys] d = (nodes[0] - x) % f.P for i in range(1, n): k = k * d % f.P d = (nodes[i] - x) % f.P t = k * nodes[i] % f.P for j, p in enumerate(polys): u[j] = u[j] * d % f.P if i < len(p): u[j] = (u[j] + t * p[i]) % f.P factor = pow(f.P - 1, n - 1, f.P) * inv(f, n) % f.P return [v * factor % f.P for v in u] def extend_values_to_power_of_2(f, p, n): """VDAF §6.1.3.2, as the draft writes it (in place).""" x = nth_root_powers(f, n) w = [0] * n for i in range(len(p)): acc = 1 for j in range(len(p)): if i != j: acc = acc * (x[i] - x[j]) % f.P w[i] = acc for k in range(len(p), n): for i in range(k): w[i] = w[i] * (x[i] - x[k]) % f.P y_num, y_den = 0, 1 for i, v in enumerate(p): y_num = (y_num * w[i] + y_den * v) % f.P y_den = y_den * w[i] % f.P acc = 1 for j in range(k): acc = acc * (x[k] - x[j]) % f.P w[k] = acc p.append((-w[k] * y_num * inv(f, y_den)) % f.P) def wire_poly_len(calls): return next_power_of_2(1 + calls) def gadget_poly_len(degree, wire_len): return degree * (wire_len - 1) + 1 # Gadgets (Appendix A). class Mul: ARITY, DEGREE = 2, 2 def eval(self, f, inp): return inp[0] * inp[1] % f.P def eval_poly(self, f, inp_poly): return poly_mul(f, inp_poly[0], inp_poly[1]) class PolyEval: ARITY = 1 def __init__(self, p, num_calls): while len(p) > 1 and p[-1] == 0: p = p[:-1] self.p = p self.DEGREE = len(p) - 1 self.n = next_power_of_2(gadget_poly_len(self.DEGREE, wire_poly_len(num_calls))) def eval(self, f, inp): return poly_eval_mono(f, [c % f.P for c in self.p], inp[0]) def eval_poly(self, f, inp_poly): m = inv_ntt(f, inp_poly[0], len(inp_poly[0])) c = [x % f.P for x in self.p] return [poly_eval_mono(f, c, x) for x in ntt(f, m, self.n)] class ParallelSum: def __init__(self, sub, count): self.sub, self.count = sub, count self.ARITY, self.DEGREE = sub.ARITY * count, sub.DEGREE def eval(self, f, inp): a = self.sub.ARITY return sum(self.sub.eval(f, inp[i * a:(i + 1) * a]) for i in range(self.count)) % f.P def eval_poly(self, f, inp_poly): out_len = next_power_of_2(gadget_poly_len(self.DEGREE, len(inp_poly[0]))) out = [0] * out_len a = self.sub.ARITY for i in range(self.count): cur = self.sub.eval_poly(f, inp_poly[i * a:(i + 1) * a]) for j in range(out_len): out[j] = (out[j] + cur[j]) % f.P return out # Validity circuits (§7.4). Each has one gadget; `g` is the gadget call. def range_encode(f, value, maxm): if value > maxm: raise ValueError("measurement is too large") bits = maxm.bit_length() rest_ones = 2 ** (bits - 1) - 1 last_weight = maxm - rest_ones rest, last = (value, 0) if value <= rest_ones else (value - last_weight, 1) return [(rest >> l) & 1 for l in range(bits - 1)] + [last] def range_decode(f, enc, maxm): bits = maxm.bit_length() last_weight = maxm - (2 ** (bits - 1) - 1) d = sum((1 << l) * b for l, b in enumerate(enc[:bits - 1])) return (d + last_weight * enc[bits - 1]) % f.P class Count: FIELD, ID = Field64, 1 def __init__(self): self.GADGET, self.CALLS = Mul(), 1 self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = 1, 0, 1, 1 def encode(self, m): if m not in (0, 1): raise ValueError("count") return [m] def eval(self, meas, jr, num_shares, g): return [(g([meas[0], meas[0]]) - meas[0]) % self.FIELD.P] def truncate(self, meas): return list(meas) class Sum: FIELD, ID = Field64, 2 def __init__(self, maxm): self.max = maxm bits = maxm.bit_length() self.GADGET, self.CALLS = PolyEval([0, -1, 1], bits), bits self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = bits, 0, 1, bits def encode(self, m): return range_encode(self.FIELD, m, self.max) def eval(self, meas, jr, num_shares, g): return [g([b]) for b in meas] def truncate(self, meas): return [range_decode(self.FIELD, meas, self.max)] def chunked_range_check(f, meas, jr, num_shares, calls, chunk, g): shares_inv = inv(f, num_shares) out = 0 for i in range(calls): r = jr[i] rp = r inputs = [] for j in range(chunk): idx = i * chunk + j m = meas[idx] if idx < len(meas) else 0 inputs += [rp * m % f.P, (m - shares_inv) % f.P] rp = rp * r % f.P out = (out + g(inputs)) % f.P return out class SumVec: FIELD, ID = Field128, 3 def __init__(self, length, maxm, chunk): self.length, self.max, self.chunk = length, maxm, chunk self.bits = maxm.bit_length() self.CALLS = (length * self.bits + chunk - 1) // chunk self.GADGET = ParallelSum(Mul(), chunk) self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = length * self.bits, self.CALLS, length, 1 def encode(self, m): if len(m) != self.length: raise ValueError("length") return [b for v in m for b in range_encode(self.FIELD, v, self.max)] def eval(self, meas, jr, num_shares, g): return [chunked_range_check(self.FIELD, meas, jr, num_shares, self.CALLS, self.chunk, g)] def truncate(self, meas): return [range_decode(self.FIELD, meas[i * self.bits:(i + 1) * self.bits], self.max) for i in range(self.length)] class Histogram: FIELD, ID = Field128, 4 def __init__(self, length, chunk): self.length, self.chunk = length, chunk self.CALLS = (length + chunk - 1) // chunk self.GADGET = ParallelSum(Mul(), chunk) self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = length, self.CALLS, length, 2 def encode(self, m): if not 0 <= m < self.length: raise ValueError("bucket") v = [0] * self.length v[m] = 1 return v def eval(self, meas, jr, num_shares, g): f = self.FIELD rc = chunked_range_check(f, meas, jr, num_shares, self.CALLS, self.chunk, g) sc = (sum(meas) - inv(f, num_shares)) % f.P return [rc, sc] def truncate(self, meas): return list(meas) # The FLP (§7.3). def flp_lens(v): g = v.GADGET p = wire_poly_len(v.CALLS) return { "prove_rand": g.ARITY, "query_rand": 1 + (v.EVAL_OUTPUT_LEN if v.EVAL_OUTPUT_LEN > 1 else 0), "proof": g.ARITY + gadget_poly_len(g.DEGREE, p), "verifier": 1 + g.ARITY + 1, } def prove(v, meas, prove_rand, joint_rand): f, g = v.FIELD, v.GADGET p = wire_poly_len(v.CALLS) wires = [[prove_rand[j]] + [0] * (p - 1) for j in range(g.ARITY)] k = [0] def call(inp): k[0] += 1 for j, x in enumerate(inp): wires[j][k[0]] = x return g.eval(f, inp) v.eval(meas, joint_rand, 1, call) return prove_rand[:g.ARITY] + g.eval_poly(f, wires)[:gadget_poly_len(g.DEGREE, p)] def query(v, meas, proof, query_rand, joint_rand, num_shares): f, g = v.FIELD, v.GADGET p = wire_poly_len(v.CALLS) gp_len = gadget_poly_len(g.DEGREE, p) seeds, poly = proof[:g.ARITY], list(proof[g.ARITY:g.ARITY + gp_len]) extend_values_to_power_of_2(f, poly, next_power_of_2(len(poly))) size = next_power_of_2(gp_len) while len(poly) < size: poly = double_evaluations(f, poly) step = size // p wires = [[s] + [0] * (p - 1) for s in seeds] k = [0] def call(inp): k[0] += 1 for j, x in enumerate(inp): wires[j][k[0]] = x return poly[k[0] * step] out = v.eval(meas, joint_rand, num_shares, call) if v.EVAL_OUTPUT_LEN > 1: r = query_rand[:v.EVAL_OUTPUT_LEN] val = sum(a * b for a, b in zip(r, out)) % f.P t = query_rand[v.EVAL_OUTPUT_LEN] else: val, t = out[0], query_rand[0] if pow(t, p, f.P) == 1: raise ValueError("test point is a root of unity") return [val] + poly_eval_batched(f, wires, t) + poly_eval_batched(f, [poly], t) def decide(v, verifier): if verifier[0] != 0: return False a = v.GADGET.ARITY return v.GADGET.eval(v.FIELD, verifier[1:1 + a]) == verifier[1 + a] # Prio3 (§7.2), one proof, any number of shares. class Prio3: def __init__(self, valid, shares=2): self.v, self.shares, self.f = valid, shares, valid.FIELD self.L = flp_lens(valid) self.jr = valid.JOINT_RAND_LEN > 0 self.rand_size = 32 * shares * (2 if self.jr else 1) def dst(self, usage, ctx): return format_dst(0, self.v.ID, usage) + ctx def helper_meas_share(self, ctx, agg_id, share): return Xof.expand_into_vec(self.f, share, self.dst(1, ctx), bytes([agg_id]), self.v.MEAS_LEN) def helper_proofs_share(self, ctx, agg_id, share): return Xof.expand_into_vec(self.f, share, self.dst(2, ctx), bytes([1, agg_id]), self.L["proof"]) def joint_rand_part(self, ctx, agg_id, blind, meas_share, nonce): return Xof.derive_seed(blind, self.dst(7, ctx), bytes([agg_id]) + nonce + encode_vec(self.f, meas_share)) def joint_rand_seed(self, ctx, parts): return Xof.derive_seed(bytes(32), self.dst(6, ctx), b"".join(parts)) def joint_rands(self, ctx, seed): return Xof.expand_into_vec(self.f, seed, self.dst(3, ctx), bytes([1]), self.v.JOINT_RAND_LEN) def shard(self, ctx, meas, nonce, rand): """shard (§7.2.1) of an encoded measurement: (public share, input shares), encoded.""" seeds = [rand[i:i + 32] for i in range(0, len(rand), 32)] n = self.shares if self.jr: hs, hb = [seeds[2 * i] for i in range(n - 1)], [seeds[2 * i + 1] for i in range(n - 1)] lb, ps = seeds[2 * (n - 1)], seeds[2 * (n - 1) + 1] else: hs, hb, lb, ps = seeds[:n - 1], [None] * (n - 1), None, seeds[n - 1] lm, parts = list(meas), [] for j in range(n - 1): hm = self.helper_meas_share(ctx, j + 1, hs[j]) lm = vsub(self.f, lm, hm) if self.jr: parts.append(self.joint_rand_part(ctx, j + 1, hb[j], hm, nonce)) jrs = [] if self.jr: parts.insert(0, self.joint_rand_part(ctx, 0, lb, lm, nonce)) jrs = self.joint_rands(ctx, self.joint_rand_seed(ctx, parts)) pr = Xof.expand_into_vec(self.f, ps, self.dst(4, ctx), bytes([1]), self.L["prove_rand"]) proof = prove(self.v, meas, pr, jrs) for j in range(n - 1): proof = vsub(self.f, proof, self.helper_proofs_share(ctx, j + 1, hs[j])) shares = [encode_vec(self.f, lm) + encode_vec(self.f, proof) + (lb if self.jr else b"")] shares += [hs[j] + (hb[j] if self.jr else b"") for j in range(n - 1)] return (b"".join(parts) if self.jr else b""), shares def decode_input_share(self, agg_id, b): bl = 32 if self.jr else 0 if agg_id == 0: ml, pl = self.v.MEAS_LEN * self.f.SIZE, self.L["proof"] * self.f.SIZE if len(b) != ml + pl + bl: raise ValueError("input share") return decode_vec(self.f, b[:ml]), decode_vec(self.f, b[ml:ml + pl]), b[ml + pl:] or None if len(b) != 32 + bl: raise ValueError("input share") return b[:32], b[32:] or None def meas_share(self, ctx, agg_id, input_share): s = self.decode_input_share(agg_id, input_share) return s[0] if agg_id == 0 else self.helper_meas_share(ctx, agg_id, s[0]) def verify_init(self, vk, ctx, agg_id, nonce, public, input_share): """verify_init (§7.2.2): (state, encoded verifier share).""" s = self.decode_input_share(agg_id, input_share) if agg_id == 0: meas, proofs, blind = s else: meas, proofs, blind = self.helper_meas_share(ctx, agg_id, s[0]), self.helper_proofs_share(ctx, agg_id, s[0]), s[1] out = self.v.truncate(meas) jrs, corrected, part = [], None, None if self.jr: if len(public) != 32 * self.shares: raise ValueError("public share") parts = [public[i:i + 32] for i in range(0, len(public), 32)] part = self.joint_rand_part(ctx, agg_id, blind, meas, nonce) parts[agg_id] = part corrected = self.joint_rand_seed(ctx, parts) jrs = self.joint_rands(ctx, corrected) elif public: raise ValueError("public share") qr = Xof.expand_into_vec(self.f, vk, self.dst(5, ctx), bytes([1]) + nonce, self.L["query_rand"]) ver = query(self.v, meas, proofs, qr, jrs, self.shares) return (out, corrected), encode_vec(self.f, ver) + (part if self.jr else b"") def message(self, ctx, shares): """verifier_shares_to_message (§7.2.2): the encoded verifier message.""" vl = self.L["verifier"] * self.f.SIZE total = [0] * self.L["verifier"] parts = [] for s in shares: if len(s) != vl + (32 if self.jr else 0): raise ValueError("verifier share") total = vadd(self.f, total, decode_vec(self.f, s[:vl])) parts.append(s[vl:]) if "proof" not in SKIP and not decide(self.v, total): raise ValueError("proof verifier check failed") return self.joint_rand_seed(ctx, parts) if self.jr else b"" def verify_next(self, state, msg): out, corrected = state if (msg if self.jr else None) != corrected or (not self.jr and msg): raise ValueError("joint randomness check failed") return encode_vec(self.f, out) def agg(self, shares): acc = [0] * self.v.OUTPUT_LEN for s in shares: v = decode_vec(self.f, s) if len(v) != self.v.OUTPUT_LEN: raise ValueError("aggregate share") acc = vadd(self.f, acc, v) return acc def prio3_of(name, doc): s = doc["shares"] if name.startswith("Prio3Count_"): return Prio3(Count(), s) if name.startswith("Prio3SumVec_"): return Prio3(SumVec(doc["length"], doc["max_measurement"], doc["chunk_length"]), s) if name.startswith("Prio3Sum_"): return Prio3(Sum(doc["max_measurement"]), s) if name.startswith("Prio3Histogram_"): return Prio3(Histogram(doc["length"], doc["chunk_length"]), s) return None def run_vdaf_file(name, doc): """One of the draft's vector files (Appendix C.1): the failures.""" fails = [] p = prio3_of(name, doc) if p is None: return ["%s: not a variant this verifier implements" % name] ctx, vk = bytes.fromhex(doc["ctx"]), bytes.fromhex(doc["verify_key"]) reps = doc["reports"] states = {} for op in doc["operations"]: kind, ok = op["operation"], op["success"] i, j = op.get("report_index"), op.get("aggregator_id") try: if kind == "shard": r = reps[i] ps, iss = p.shard(ctx, p.v.encode(r["measurement"]), bytes.fromhex(r["nonce"]), bytes.fromhex(r["rand"])) got = ps.hex() == r["public_share"] and [x.hex() for x in iss] == r["input_shares"] elif kind == "verify_init": r = reps[i] st, vs = p.verify_init(vk, ctx, j, bytes.fromhex(r["nonce"]), bytes.fromhex(r["public_share"]), bytes.fromhex(r["input_shares"][j])) states[(i, j)] = st got = vs.hex() == r["verifier_shares"][0][j] elif kind == "verifier_shares_to_message": r = reps[i] got = p.message(ctx, [bytes.fromhex(x) for x in r["verifier_shares"][0]]).hex() == r["verifier_messages"][0] elif kind == "verify_next": r = reps[i] got = p.verify_next(states[(i, j)], bytes.fromhex(r["verifier_messages"][0])).hex() == r["out_shares"][j] elif kind == "aggregate": got = encode_vec(p.f, p.agg([bytes.fromhex(r["out_shares"][j]) for r in reps])).hex() == doc["agg_shares"][j] elif kind == "unshard": total = p.agg([bytes.fromhex(x) for x in doc["agg_shares"]]) res = total[0] if isinstance(doc["agg_result"], int) else total got = res == doc["agg_result"] else: got = False if not (ok and got): fails.append("%s: %s %s %s gives other bytes" % (name, kind, i, j)) except (ValueError, KeyError, IndexError): if ok: fails.append("%s: %s %s %s refused" % (name, kind, i, j)) return fails def run_xof(doc): seed, dst, binder = (bytes.fromhex(doc[k]) for k in ("seed", "dst", "binder")) fails = [] if Xof.derive_seed(seed, dst, binder).hex() != doc["derived_seed"]: fails.append("XofTurboShake128: derived seed") if encode_vec(Field128, Xof.expand_into_vec(Field128, seed, dst, binder, doc["length"])).hex() != doc["expanded_vec_field128"]: fails.append("XofTurboShake128: expansion") return fails # --------------------------------------------------------------------------------------- # X25519 (RFC 7748 §5), HKDF-SHA256 (RFC 5869), AES-128 (FIPS 197), GCM (SP 800-38D), HPKE # base mode (RFC 9180). P25519 = 2 ** 255 - 19 def x25519(k, u): k = bytearray(k) k[0] &= 248 k[31] &= 127 k[31] |= 64 k = int.from_bytes(k, "little") x1 = int.from_bytes(u, "little") & ((1 << 255) - 1) x2, z2, x3, z3, swap = 1, 0, x1, 1, 0 for t in reversed(range(255)): kt = (k >> t) & 1 swap ^= kt if swap: x2, x3, z2, z3 = x3, x2, z3, z2 swap = kt a, b, c, d = x2 + z2, x2 - z2, x3 + z3, x3 - z3 aa, bb = a * a % P25519, b * b % P25519 e = (aa - bb) % P25519 da, cb = d * a % P25519, c * b % P25519 x3 = (da + cb) ** 2 % P25519 z3 = x1 * (da - cb) ** 2 % P25519 x2 = aa * bb % P25519 z2 = e * (aa + 121665 * e) % P25519 if swap: x2, z2 = x3, z3 return (x2 * pow(z2, P25519 - 2, P25519) % P25519).to_bytes(32, "little") BASE = (9).to_bytes(32, "little") def hkdf_extract(salt, ikm): return hmac.new(salt, ikm, hashlib.sha256).digest() def hkdf_expand(prk, info, n): out, t, i = b"", b"", 1 while len(out) < n: t = hmac.new(prk, t + info + bytes([i]), hashlib.sha256).digest() out += t i += 1 return out[:n] def _xt(a): return ((a << 1) ^ 0x1B) & 0xFF if a & 0x80 else a << 1 def _mul(a, b): r = 0 while b: if b & 1: r ^= a a, b = _xt(a), b >> 1 return r def _sbox(): inv_ = [0] * 256 for a in range(1, 256): for b in range(1, 256): if _mul(a, b) == 1: inv_[a] = b break s = [] for a in range(256): x = inv_[a] y = x for _ in range(4): x = ((x << 1) | (x >> 7)) & 0xFF y ^= x s.append(y ^ 0x63) return s SBOX = _sbox() def aes_keys(key): w = [list(key[4 * i:4 * i + 4]) for i in range(4)] rcon = 1 for i in range(4, 44): t = list(w[i - 1]) if i % 4 == 0: t = [SBOX[x] for x in t[1:] + t[:1]] t[0] ^= rcon rcon = _xt(rcon) w.append([a ^ b for a, b in zip(w[i - 4], t)]) return [sum(w[4 * r:4 * r + 4], []) for r in range(11)] def aes_block(rk, blk): s = [a ^ b for a, b in zip(blk, rk[0])] for r in range(1, 11): s = [SBOX[x] for x in s] s = [s[(i + 4 * (i % 4)) % 16] for i in range(16)] # ShiftRows, column-major state if r < 10: m = [] for c in range(4): a = s[4 * c:4 * c + 4] m += [_mul(a[0], 2) ^ _mul(a[1], 3) ^ a[2] ^ a[3], a[0] ^ _mul(a[1], 2) ^ _mul(a[2], 3) ^ a[3], a[0] ^ a[1] ^ _mul(a[2], 2) ^ _mul(a[3], 3), _mul(a[0], 3) ^ a[1] ^ a[2] ^ _mul(a[3], 2)] s = m s = [a ^ b for a, b in zip(s, rk[r])] return bytes(s) R_GCM = 0xE1 << 120 def gf_mul(x, y): z, v = 0, y for i in range(127, -1, -1): if (x >> i) & 1: z ^= v v = (v >> 1) ^ R_GCM if v & 1 else v >> 1 return z def ghash(h, data): y = 0 for i in range(0, len(data), 16): y = gf_mul(y ^ int.from_bytes(data[i:i + 16], "big"), h) return y def _pad16(b): return b + bytes((-len(b)) % 16) def gcm_open(key, nonce, aad, ct): if len(ct) < 16: raise ValueError("short") rk = aes_keys(key) h = int.from_bytes(aes_block(rk, bytes(16)), "big") j0 = nonce + b"\x00\x00\x00\x01" c, tag = ct[:-16], ct[-16:] s = ghash(h, _pad16(aad) + _pad16(c) + (8 * len(aad)).to_bytes(8, "big") + (8 * len(c)).to_bytes(8, "big")) want = (int.from_bytes(aes_block(rk, j0), "big") ^ s).to_bytes(16, "big") if not hmac.compare_digest(want, tag): raise ValueError("tag") out = bytearray() ctr = int.from_bytes(j0[12:], "big") for i in range(0, len(c), 16): ctr = (ctr + 1) & 0xFFFFFFFF ks = aes_block(rk, nonce + ctr.to_bytes(4, "big")) out += bytes(a ^ b for a, b in zip(c[i:i + 16], ks)) return bytes(out) KEM_SUITE = b"KEM" + (0x0020).to_bytes(2, "big") HPKE_SUITE = b"HPKE" + (0x0020).to_bytes(2, "big") + (1).to_bytes(2, "big") + (1).to_bytes(2, "big") def labeled_extract(suite, salt, label, ikm): return hkdf_extract(salt, b"HPKE-v1" + suite + label + ikm) def labeled_expand(suite, prk, label, info, n): return hkdf_expand(prk, n.to_bytes(2, "big") + b"HPKE-v1" + suite + label + info, n) def hpke_open(enc, sk, info, aad, ct): dh = x25519(sk, enc) if dh == bytes(32): raise ValueError("zero") ctx = enc + x25519(sk, BASE) shared = labeled_expand(KEM_SUITE, labeled_extract(KEM_SUITE, b"", b"eae_prk", dh), b"shared_secret", ctx, 32) ksc = b"\x00" + labeled_extract(HPKE_SUITE, b"", b"psk_id_hash", b"") + labeled_extract(HPKE_SUITE, b"", b"info_hash", info) secret = labeled_extract(HPKE_SUITE, shared, b"secret", b"") key = labeled_expand(HPKE_SUITE, secret, b"key", ksc, 16) nonce = labeled_expand(HPKE_SUITE, secret, b"base_nonce", ksc, 12) return gcm_open(key, nonce, aad, ct) def crypto_self_test(): """RFC 9861 §5's TurboSHAKE128 vectors, RFC 9180 A.1.1 (DeriveKeyPair, OpenBase), RFC 7748 §5.2's first X25519 vector: the checks that fail.""" fails = [] if turboshake128(b"", 0x1F, 32).hex() != "1e415f1c5983aff2169217277d17bb538cd945a397ddec541f1ce41af2c1b74c": fails.append("TurboSHAKE128 empty") if turboshake128(bytes(i % 251 for i in range(17 ** 3)), 0x1F, 32).hex() != "d4976eb56bcf118520582b709f73e1d6853e001fdaf80e1b13e0d0599d5fb372": fails.append("TurboSHAKE128 ptn(17^3)") if turboshake128(b"\xff" * 3, 0x01, 32).hex() != "bf323f940494e88ee1c540fe660be8a0c93f43d15ec006998462fa994eed5dab": fails.append("TurboSHAKE128 D=01") k = bytes.fromhex("a546e36bf0527c9d3b16154b82465edd62144c0ac1fc5a18506a2244ba449ac4") u = bytes.fromhex("e6db6867583030db3594c1a424b15f7c726624ec26b3353b10a903a6d0ab1c4c") if x25519(k, u).hex() != "c3da55379de9c6908e94ea4df28d084f32eccf03491c71f754b4075577a28552": fails.append("X25519") ikm_r = bytes.fromhex("6db9df30aa07dd42ee5e8181afdb977e538f5e1fec8a06223f33f7013e525037") sk_r = labeled_expand(KEM_SUITE, labeled_extract(KEM_SUITE, b"", b"dkp_prk", ikm_r), b"sk", b"", 32) if sk_r.hex() != "4612c550263fc8ad58375df3f557aac531d26850903e55a9f23f21d8534e8ac8": fails.append("HPKE DeriveKeyPair") try: pt = hpke_open(bytes.fromhex("37fda3567bdbd628e88668c3c8d7e97d1d1253b6d4ea6d44c150f741f1bf4431"), sk_r, bytes.fromhex("4f6465206f6e2061204772656369616e2055726e"), bytes.fromhex("436f756e742d30"), bytes.fromhex("f938558b5d72f1a23810b4be2ab4f84331acc02fc97babc53a52ae8218a355a96d8770ac83d07bea87e13c512a")) if pt.hex() != "4265617574792069732074727574682c20747275746820626561757479": fails.append("HPKE OpenBase plaintext") except ValueError: fails.append("HPKE OpenBase") return fails # --------------------------------------------------------------------------------------- # DAP's messages (draft-ietf-ppm-dap-20 §4), read strictly. class Rd: def __init__(self, b): self.b, self.i = b, 0 def take(self, n): if self.i + n > len(self.b): raise ValueError("truncated") x = self.b[self.i:self.i + n] self.i += n return x def u(self, n): return int.from_bytes(self.take(n), "big") def opaque(self, n): return self.take(self.u(n)) def done(self): return self.i == len(self.b) def end(self): if not self.done(): raise ValueError("trailing") def read_exts(r): body = Rd(r.opaque(2)) out = [] while not body.done(): out.append((body.u(2), body.opaque(2))) return out def read_task(r): t = {"task_info": r.opaque(1), "leader": r.opaque(2), "helper": r.opaque(2), "time_precision": r.u(8), "min_batch_size": r.u(8), "batch_mode": r.u(1), "batch_config": r.opaque(2), "vdaf_type": r.u(4), "vdaf_config": r.opaque(2), "extensions": read_exts(r)} if not t["leader"] or not t["helper"]: raise ValueError("empty URL") return t def read_ct(r): c = {"config_id": r.u(1), "enc": r.opaque(2), "payload": r.opaque(4)} if not c["enc"] or not c["payload"]: raise ValueError("empty") return c def write_ct(c): return bytes([c["config_id"]]) + len(c["enc"]).to_bytes(2, "big") + c["enc"] + len(c["payload"]).to_bytes(4, "big") + c["payload"] def read_metadata(r): start = r.i m = {"id": r.take(16), "time": r.u(8), "exts": read_exts(r)} m["raw"] = r.b[start:r.i] return m def read_report(r): m = read_metadata(r) return {"m": m, "public": r.opaque(4), "leader": read_ct(r), "helper": read_ct(r)} def read_collection(b): r = Rd(b) c = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4), "extensions": read_exts(r)} r.end() return c def interval_of(b): r = Rd(b) i = (r.u(8), r.u(8)) r.end() return i def within(a, b): return a[0] >= b[0] and a[0] + a[1] <= b[0] + b[1] def u64(x): return x.to_bytes(8, "big") # --------------------------------------------------------------------------------------- # The extension: measures (§3), tasks (§4), statements (§6), receipts (§7), verification (§8), # cells (§9.2). FLOOR = 50 MEASURES = {1: "outcomes", 2: "stalls", 3: "durations", 4: "start_time"} START_EDGES = [250, 500, 1000, 2000, 4000, 8000] def measure_vdaf(code): return {1: lambda: Prio3(SumVec(4, 1, 2)), 2: lambda: Prio3(Sum(65535)), 3: lambda: Prio3(SumVec(3, 4294967295, 10)), 4: lambda: Prio3(Histogram(8, 3))}[code]() def dap_config(p): v = p.v if isinstance(v, Count): return b"" if isinstance(v, Sum): return u64(v.max) if isinstance(v, SumVec): return v.length.to_bytes(4, "big") + u64(v.max) + v.chunk.to_bytes(4, "big") return v.length.to_bytes(4, "big") + v.chunk.to_bytes(4, "big") def measurement(code, m): """§3.1: a session's measurement, before the VDAF encodes it.""" if code == 1: return [int(m["started"]), int(m["start_failed"]), int(m["exit_before_start"]), int(m["vst_ms"] is not None)] if code == 2: return [min(m["stalls"], 65535)] if code == 3: return [min(x, 4294967295) for x in (m["stall_ms"], m["engaged_ms"], m["vst_ms"] or 0)] v = m["vst_ms"] if v is None: return [7] return [next((i for i, e in enumerate(START_EDGES) if v < e), 6)] def vdaf_encode(code, p, meas): return p.v.encode(meas if code in (1, 3) else meas[0]) def task_info(b): if len(b) != 153 or b[:8] != b"wai-qoe1": return None return {"measure": b[8], "scope": b[9:41], "salt": b[41:57], "leader": b[57:89], "helper": b[89:121], "collector": b[121:153]} def task_id(cfg): return sha256(b"wai:telemetry-aggregate/task\x00" + cfg) def check_task(cfg_bytes): """§4.3, in order: the task's fields, its info and its Prio3.""" try: r = Rd(cfg_bytes) t = read_task(r) r.end() except ValueError: raise Refuse("task_invalid", "task") info = task_info(t["task_info"]) if "task-rules" not in SKIP: if info is None or t["time_precision"] == 0 or t["batch_mode"] != 1 or t["batch_config"]: raise Refuse("task_invalid", "task") types = [e[0] for e in t["extensions"]] if any(e[0] != 1 or len(e[1]) != 16 for e in t["extensions"]) or types != sorted(set(types)): raise Refuse("task_invalid", "task") if info is None or info["measure"] not in MEASURES: raise Refuse("measure_unknown", "task") p = measure_vdaf(info["measure"]) if "measure" not in SKIP and (t["vdaf_type"] != p.v.ID or t["vdaf_config"] != dap_config(p)): raise Refuse("measure_mismatch", "task") if "floor-rule" not in SKIP and t["min_batch_size"] < FLOOR: raise Refuse("floor_too_low", "task") if "one-key" not in SKIP and info["leader"] == info["helper"]: raise Refuse("aggregator_alone", "task") return t, info, p, task_id(cfg_bytes) def read_statement(b): try: r = Rd(b) if r.take(4) != b"WQS1" or r.u(1) != 1: raise ValueError("magic") s = {"role": r.u(1), "task_id": r.take(32), "collection": r.take(32), "batch": (r.u(8), r.u(8)), "count": r.u(8), "checksum": r.take(32), "agg_share": r.opaque(4), "signer": r.take(32)} s["body"] = b[:r.i] s["sig"] = r.take(64) r.end() except ValueError: raise Refuse("statement_invalid") if s["role"] not in (2, 3): raise Refuse("statement_invalid") return s def read_receipt(b): r = Rd(b) try: if r.take(4) != b"WQA1": raise Refuse("malformed", "receipt") if r.u(1) != 1: raise Refuse("unknown_version", "receipt") rc = {"task": r.opaque(2), "collection": r.opaque(4), "interval": (r.u(8), r.u(8)), "leader": r.opaque(4), "helper": r.opaque(4)} rc["result"] = [int.from_bytes(r.take(16), "big") for _ in range(r.u(2))] rc["signer"] = r.take(32) rc["body"] = b[:r.i] rc["sig"] = r.take(64) except ValueError: raise Refuse("truncated", "receipt") if not r.done(): raise Refuse("malformed", "receipt") return rc def sig_check(key, domain, body, sig, at): e = ED.verify(key, domain + body, sig) if e is not None: raise Refuse(e, at) def verify_receipt(b, reports): """§8.1, in order.""" rc = read_receipt(b) t, info, p, tid = check_task(rc["task"]) if "collector-signer" not in SKIP and rc["signer"] != info["collector"]: raise Refuse("wrong_signer", "receipt") if "collector-sig" not in SKIP: sig_check(rc["signer"], b"wai:telemetry-aggregate/receipt\x01", rc["body"], rc["sig"], "receipt") try: c = read_collection(rc["collection"]) q = interval_of(c["config"]) if c["batch_mode"] == 1 else None except ValueError: raise Refuse("collection_invalid", "collection") if "collection" not in SKIP and (q is None or q[1] == 0 or c["agg_param"] or c["extensions"]): raise Refuse("collection_invalid", "collection") if not rc["leader"] or not rc["helper"]: raise Refuse("aggregator_alone", "statements") sts = [] for raw, at in ((rc["leader"], "leader"), (rc["helper"], "helper")): try: sts.append(read_statement(raw)) except Refuse as e: raise Refuse(e.reason, at) l, h = sts if "two-aggregators" not in SKIP and (l["signer"] == h["signer"] or l["role"] == h["role"]): raise Refuse("aggregator_alone", "statements") for s, role, key, at in ((l, 2, info["leader"], "leader"), (h, 3, info["helper"], "helper")): if "statement-signer" not in SKIP and (s["role"] != role or s["signer"] != key): raise Refuse("wrong_signer", at) if "statement-sig" not in SKIP: sig_check(s["signer"], b"wai:telemetry-aggregate/share\x01", s["body"], s["sig"], at) for s, at in ((l, "leader"), (h, "helper")): if "binding" not in SKIP and (s["task_id"] != tid or s["collection"] != sha256(rc["collection"])): raise Refuse("task_mismatch", at) for s, at in ((l, "leader"), (h, "helper")): if "batch" not in SKIP and (s["batch"][1] == 0 or not within(s["batch"], q)): raise Refuse("batch_invalid", at) if "batch" not in SKIP and (rc["interval"][1] == 0 or not within(rc["interval"], l["batch"])): raise Refuse("batch_invalid", "interval") if "agreement" not in SKIP and (l["batch"], l["count"], l["checksum"]) != (h["batch"], h["count"], h["checksum"]): raise Refuse("statements_disagree", "statements") if "floor" not in SKIP and l["count"] < t["min_batch_size"]: raise Refuse("batch_too_small", "statements") shares = [] for s, at in ((l, "leader"), (h, "helper")): try: shares.append(p.agg([s["agg_share"]])) except ValueError: if "share" not in SKIP: raise Refuse("share_invalid", at) shares.append([0] * p.v.OUTPUT_LEN) result = vadd(p.f, shares[0], shares[1]) if "result" not in SKIP and result != rc["result"]: raise Refuse("result_mismatch", "result") if reports is not None and "reports" not in SKIP: ids, ck = set(), bytes(32) for g in reports: gid = bytes.fromhex(g["id"]) if bytes.fromhex(g["task"]) != tid or not rc["interval"][0] <= g["time"] < sum(rc["interval"]) or gid in ids: raise Refuse("batch_mismatch", "reports") ids.add(gid) ck = bytes(a ^ b for a, b in zip(ck, sha256(gid))) if len(reports) != l["count"] or ck != l["checksum"]: raise Refuse("batch_mismatch", "reports") return { "outcome": "accept", "digest": sha256(rc["body"]).hex(), "task_id": tid.hex(), "measure": MEASURES[info["measure"]], "vdaf": {1: "Prio3Count", 2: "Prio3Sum", 3: "Prio3SumVec", 4: "Prio3Histogram"}[p.v.ID], "vdaf_config": t["vdaf_config"].hex(), "scope": info["scope"].hex(), "time_precision": t["time_precision"], "min_batch_size": t["min_batch_size"], "batch": {"start": l["batch"][0], "duration": l["batch"][1]}, "interval": {"start": rc["interval"][0], "duration": rc["interval"][1]}, "report_count": l["count"], "result": rc["result"], "leader": info["leader"].hex(), "helper": info["helper"].hex(), "collector": info["collector"].hex(), } def receipt_outcome(v): try: return verify_receipt(bytes.fromhex(v["receipt"]), v["reports"]) except Refuse as e: return {"outcome": "refuse", "reason": e.reason, "at": e.at} CELL_FIELDS = [("started", "outcomes", 0), ("start_failed", "outcomes", 1), ("exit_before_start", "outcomes", 2), ("stalls", "stalls", 0), ("stall_ms", "durations", 0), ("engaged_ms", "durations", 1), ("vst_count", "outcomes", 3), ("vst_ms", "durations", 2)] def cell_outcome(doc, v): acc = [] for n in v["receipts"]: r = next((x for x in doc["receipts"] if x["name"] == n), None) if r is None: return {"outcome": "refuse", "reason": "cell_mismatch"} o = receipt_outcome(r) if o["outcome"] != "accept": return {"outcome": "refuse", "reason": o["reason"]} acc.append(o) by = {} for a in acc: by.setdefault(a["measure"], []).append(a) if "cell" not in SKIP: if len(acc) != 3 or sorted(by) != ["durations", "outcomes", "stalls"] or any(len(x) != 1 for x in by.values()): return {"outcome": "refuse", "reason": "cell_mismatch"} if len({a["scope"] for a in acc}) != 1 or acc[0]["scope"] == "00" * 32 or len({json.dumps(a["batch"]) for a in acc}) != 1: return {"outcome": "refuse", "reason": "cell_mismatch"} def pick(m): return (by.get(m) or acc)[0] try: return {"outcome": "accept", "scope": pick("outcomes")["scope"], "batch": pick("outcomes")["batch"], "counts": [pick(m)["report_count"] for m in ("outcomes", "stalls", "durations")], "cell": {f: pick(m)["result"][i] for f, m, i in CELL_FIELDS}} except IndexError: return {"outcome": "refuse", "reason": "cell_mismatch"} def plaintext_cell(sessions): """live-metering §4.7 over the sessions' metrics.""" ms = [s["metrics"] for s in sessions] return {"started": sum(m["started"] for m in ms), "start_failed": sum(m["start_failed"] for m in ms), "exit_before_start": sum(m["exit_before_start"] for m in ms), "stalls": sum(m["stalls"] for m in ms), "stall_ms": sum(m["stall_ms"] for m in ms), "engaged_ms": sum(m["engaged_ms"] for m in ms), "vst_count": sum(m["vst_ms"] is not None for m in ms), "vst_ms": sum(m["vst_ms"] or 0 for m in ms)} # --------------------------------------------------------------------------------------- # A flow, re-run (§5, DAP §4.4 to §4.6). def input_share_aad(tid, cfg, m, public): return tid + cfg + m["raw"] + len(public).to_bytes(4, "big") + public def agg_share_aad(tid, cfg, coll): return tid + cfg + coll def open_share(sk, role, aad, ct): pt = hpke_open(ct["enc"], sk, b"dap-20 input share" + bytes([1, role]), aad, ct["payload"]) r = Rd(pt) exts = read_exts(r) payload = r.opaque(4) r.end() if exts or not payload: raise ValueError("extensions") return payload def check_flow(doc, f, fails, quiet): name = f["name"] K = doc["keys"] def bad(what): fails.append("flow %s: %s" % (name, what)) cfg = bytes.fromhex(f["task"]) try: t, info, p, tid = check_task(cfg) except Refuse as e: return bad("its task is refused %s" % e.reason) code = info["measure"] if tid.hex() != f["task_id"] or MEASURES[code] != name: bad("task id or measure") vk = hkdf_expand(hkdf_extract(b"verify_key", bytes.fromhex(K["verify_key_seed"])), tid, 32) if vk.hex() != f["verify_key"]: bad("verification key (DAP §8.6.2)") ctx = b"dap-20" + tid lsk, hsk, csk = (bytes.fromhex(K[k]["private_key"]) for k in ("leader_hpke", "helper_hpke", "collector_hpke")) cfgs = {} for k, sk in (("leader_hpke", lsk), ("helper_hpke", hsk), ("collector_hpke", csk)): r = Rd(bytes.fromhex(K[k]["config"])) c = {"id": r.u(1), "kem": r.u(2), "kdf": r.u(2), "aead": r.u(2), "pk": r.opaque(2)} r.end() if (c["kem"], c["kdf"], c["aead"]) != (0x20, 1, 1) or c["pk"] != x25519(sk, BASE): bad("%s is not DAP §7's suite with its key" % k) cfgs[k] = c sessions = {s["session"]: s for s in doc["sessions"]} up = Rd(bytes.fromhex(f["upload"])) reports = [] try: while not up.done(): reports.append(read_report(up)) except ValueError: return bad("the upload does not read") listed = f["reports"] if [r["m"]["id"].hex() for r in reports] != [x["id"] for x in listed] or [r["m"]["time"] for r in reports] != [x["time"] for x in listed]: return bad("the upload is not the listed reports") seen, pending, upload_errors = set(), [], [] for r, x in zip(reports, listed): if r["leader"]["config_id"] != cfgs["leader_hpke"]["id"]: upload_errors.append({"id": x["id"], "error": "hpke_unknown_config_id"}) elif r["m"]["id"] in seen: upload_errors.append({"id": x["id"], "error": "report_replayed"}) else: seen.add(r["m"]["id"]) pending.append((r, x)) if upload_errors != f["upload_errors"]: bad("the reports the leader refused at upload") init = bytes([0]) + (0).to_bytes(4, "big") + (0).to_bytes(2, "big") resp, outcomes, committed = b"", [], [] for r, x in pending: m = r["m"] laad = input_share_aad(tid, cfg, m, r["public"]) try: lpay = open_share(lsk, 2, laad, r["leader"]) except ValueError: outcomes.append({"id": x["id"], "outcome": "hpke_decrypt_error"}) continue try: lstate, lvs = p.verify_init(vk, ctx, 0, m["id"], r["public"], lpay) except ValueError: outcomes.append({"id": x["id"], "outcome": "vdaf_verify_error"}) continue outbound = b"\x00" + len(lvs).to_bytes(4, "big") + lvs init += m["raw"] + len(r["public"]).to_bytes(4, "big") + r["public"] + write_ct(r["helper"]) + len(outbound).to_bytes(4, "big") + outbound try: hpay = open_share(hsk, 3, laad, r["helper"]) except ValueError: resp += m["id"] + b"\x02" + bytes([5]) outcomes.append({"id": x["id"], "outcome": "hpke_decrypt_error"}) if x["kind"] != "corrupt": bad("report %s does not open" % x["id"]) continue if True: enc = vadd(p.f, p.meas_share(ctx, 0, lpay), p.meas_share(ctx, 1, hpay)) if x["kind"] in ("session",): meas = measurement(code, sessions[x["session"]]["metrics"]) if meas != x["measurement"] or enc != vdaf_encode(code, p, meas): bad("report %s is not its session's measurement (§3.1)" % x["id"]) elif x["kind"] == "out_of_range": if enc != x["encoded"]: bad("the out-of-range report's encoding") try: hstate, hvs = p.verify_init(vk, ctx, 1, m["id"], r["public"], hpay) msg = p.message(ctx, [lvs, hvs]) hout = p.verify_next(hstate, msg) except ValueError: resp += m["id"] + b"\x02" + bytes([6]) outcomes.append({"id": x["id"], "outcome": "vdaf_verify_error"}) continue fin = b"\x02" + len(msg).to_bytes(4, "big") + msg resp += m["id"] + b"\x00" + len(fin).to_bytes(4, "big") + fin lout = p.verify_next(lstate, msg) committed.append((m["id"], m["time"], lout, hout)) outcomes.append({"id": x["id"], "outcome": "committed"}) if init.hex() != f["job"]["init_req"]: bad("the AggregationJobInitReq is not DAP §4.5.3.1's") if resp.hex() != f["job"]["resp"]: bad("the AggregationJobResp is not DAP §4.5.3.2's") if outcomes != f["job"]["outcomes"]: bad("the reports' outcomes") if "proof" not in SKIP: for x in listed: o = next((y["outcome"] for y in outcomes if y["id"] == x["id"]), None) if x["kind"] == "out_of_range" and o != "vdaf_verify_error": bad("the out-of-range report is not refused vdaf_verify_error") buckets = {} for rid, tm, lo, ho in committed: b = buckets.setdefault(tm, {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "ids": []}) b["l"], b["h"] = vadd(p.f, b["l"], decode_vec(p.f, lo)), vadd(p.f, b["h"], decode_vec(p.f, ho)) b["n"] += 1 b["ck"] = bytes(a ^ c for a, c in zip(b["ck"], sha256(rid))) b["ids"].append((rid, tm)) def merged(iv): out = {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "times": [], "ids": []} for tm in range(iv[0], iv[0] + iv[1]): if tm in buckets: b = buckets[tm] out["l"], out["h"] = vadd(p.f, out["l"], b["l"]), vadd(p.f, out["h"], b["h"]) out["n"] += b["n"] out["ck"] = bytes(a ^ c for a, c in zip(out["ck"], b["ck"])) out["times"].append(tm) out["ids"] += b["ids"] return out coll_raw = bytes.fromhex(f["collection"]["request"]) query = interval_of(read_collection(coll_raw)["config"]) for v in f["refusals"]: collected = query if v["phase"] == "after" else None req = bytes.fromhex(v["request"]) def overlaps(iv): return collected is not None and iv[0] < collected[0] + collected[1] and collected[0] < iv[0] + iv[1] try: if v["actor"] == "leader": # DAP §4.6.1 c = read_collection(req) q = interval_of(c["config"]) if c["agg_param"]: got = "invalidAggregationParameter" elif q[1] == 0: got = "batchInvalid" elif overlaps(q): got = "batchOverlap" elif merged(q)["n"] < t["min_batch_size"]: got = "invalidBatchSize" else: got = "accept" else: # DAP §4.6.4 r = Rd(req) c = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4), "extensions": read_exts(r)} mode, sel, n, ck = r.u(1), r.opaque(2), r.u(8), r.take(32) r.end() q, b = interval_of(c["config"]), interval_of(sel) mg = merged(b) if mode != 1 or c["batch_mode"] != 1: got = "invalidMessage" elif b[1] == 0 or not within(b, q): got = "batchInvalid" elif overlaps(b): got = "batchOverlap" elif mg["n"] < t["min_batch_size"]: got = "invalidBatchSize" elif c["agg_param"]: got = "invalidMessage" elif (n, ck) != (mg["n"], mg["ck"]): got = "batchMismatch" else: got = "accept" except ValueError: got = "invalidMessage" if got != v["outcome"]: bad("refusal %s: stated %s, DAP gives %s" % (v["name"], v["outcome"], got)) # The collection (DAP §4.6). mg = merged(query) if mg["n"] < t["min_batch_size"] and "floor" not in SKIP: bad("a batch below the minimum was collected") col = f["collection"] share_req = coll_raw + b"\x01" + (16).to_bytes(2, "big") + u64(query[0]) + u64(query[1]) + u64(mg["n"]) + mg["ck"] if share_req.hex() != col["share_req"]: bad("the AggregateShareReq is not DAP §4.6.4's") aad = agg_share_aad(tid, cfg, coll_raw) try: hct = read_ct(Rd(bytes.fromhex(col["helper_share"]))) hshare = hpke_open(hct["enc"], csk, b"dap-20 aggregate share" + bytes([3, 0]), aad, hct["payload"]) r = Rd(bytes.fromhex(col["resp"])) n, iv = r.u(8), (r.u(8), r.u(8)) lct, hct2 = read_ct(r), read_ct(r) r.end() lshare = hpke_open(lct["enc"], csk, b"dap-20 aggregate share" + bytes([2, 0]), aad, lct["payload"]) sts = [] for k, role in (("leader_statement", 2), ("helper_statement", 3)): sct = read_ct(Rd(bytes.fromhex(col[k]))) sts.append(read_statement(hpke_open(sct["enc"], csk, b"wai-telemetry-aggregate statement" + bytes([role, 0]), aad, sct["payload"]))) except (ValueError, Refuse): return bad("the collection does not open with the collector's key") times = mg["times"] if n != mg["n"] or iv != (min(times), max(times) - min(times) + 1) or hct2 != hct: bad("the CollectionJobResp's count, interval or helper share") if decode_vec(p.f, lshare) != mg["l"] or decode_vec(p.f, hshare) != mg["h"]: bad("an aggregate share is not the batch's") for s, role, share, key in ((sts[0], 2, lshare, info["leader"]), (sts[1], 3, hshare, info["helper"])): if (s["role"], s["task_id"], s["collection"], s["batch"], s["count"], s["checksum"], s["agg_share"], s["signer"]) != \ (role, tid, sha256(coll_raw), query, mg["n"], mg["ck"], share, key): bad("the %s statement is not its share's (§6.2)" % ("leader" if role == 2 else "helper")) if ED.verify(key, b"wai:telemetry-aggregate/share\x01" + s["body"], s["sig"]) is not None: bad("a statement's signature") result = vadd(p.f, mg["l"], mg["h"]) if result != f["result"]: bad("the result") rc = read_receipt(bytes.fromhex(f["receipt"])) if (rc["leader"], rc["helper"], rc["result"], rc["task"], rc["collection"], rc["interval"]) != \ (sts[0]["body"] + sts[0]["sig"], sts[1]["body"] + sts[1]["sig"], result, cfg, coll_raw, iv): bad("the receipt is not the collection's") given = [{"id": rid.hex(), "time": tm, "task": tid.hex()} for rid, tm in mg["ids"]] named = next((x for x in doc["receipts"] if x["name"] == name), None) if named is None or named["receipt"] != f["receipt"] or sorted(named["reports"] or [], key=json.dumps) != sorted(given, key=json.dumps): bad("the receipts section does not hold the flow's receipt with its reports") if not quiet: print("PASS flow %s: %d reports re-run, %d committed, result recomputed" % (name, len(reports), len(committed))) # --------------------------------------------------------------------------------------- # delivery-telemetry §3.4 and §4.3. def key_classes(): with open(TELEMETRY, encoding="utf-8") as fh: text = fh.read() at = text.index("\n§3.4") end = text.index("\n## ", at) out = {} for line in text[at:end].splitlines(): if line.startswith("| `"): cells = [c.strip() for c in line.split("|")] for k in cells[1].split(", "): out[k.strip("`")] = cells[2] return out CLASSES = key_classes() def sf_dict_members(s): """Top-level members of an RFC 9651 Dictionary as (key, raw value), keeping strings whole.""" out, i = [], 0 while i < len(s): j = i while j < len(s) and s[j] not in "=,": j += 1 key = s[i:j] if j < len(s) and s[j] == "=": k = j + 1 if s[k] == '"': k += 1 while s[k] != '"': k += 2 if s[k] == "\\" else 1 k += 1 elif s[k] == "(": k = s.index(")", k) + 1 else: while k < len(s) and s[k] != ",": k += 1 out.append((key, s[j + 1:k])) i = k else: out.append((key, "?1")) i = j if i < len(s): if s[i] != ",": raise ValueError("dictionary") i += 1 return out def shard_of_text(): with open(TELEMETRY, encoding="utf-8") as fh: text = fh.read() at = text.index("§4.2 In Request mode") para = text[at:text.index("\n\n", at)] m = {} for group, shard in re.findall(r"((?:`[a-z]+`(?:, | and )?)+) to `(CMCD-[A-Za-z]+)`", para.replace("\n", " ")): for k in re.findall(r"`([a-z]+)`", group): m[k] = shard t41 = text[text.index("§4.1 Custom keys"):text.index("§4.2 In Request mode")] for line in t41.splitlines(): if line.startswith("| `science"): cells = [c.strip() for c in line.split("|")] keys = [cells[1].split("`")[1]] rest = re.findall(r"`(-[a-z]+)`", cells[1]) base = keys[0].rsplit("-", 1)[0] keys += [base + x for x in rest] if cells[4].startswith("`CMCD"): for k in keys: m[k] = cells[4].strip("`") return m def delivery_only_outcome(v): try: mem = sf_dict_members(v["delivery_only"]) except (ValueError, IndexError): return {"outcome": "refuse", "reason": "measurement_key_in_delivery", "key": "?"} for k, _ in mem: if CLASSES.get(k) != "delivery": return {"outcome": "refuse", "reason": "measurement_key_in_delivery", "key": k} return {"outcome": "accept"} def check_delivery_only(v, fails): got = delivery_only_outcome(v) if got != v["outcome"]: fails.append("delivery-only %s: stated %s, got %s" % (v["name"], v["outcome"], got)) if "report" in v: full = sf_dict_members(v["report"]) want = ",".join("%s=%s" % kv for kv in full if CLASSES.get(kv[0]) == "delivery") if want != v["delivery_only"]: fails.append("delivery-only %s: not the report's delivery-class members" % v["name"]) shards = {} sm = shard_of_text() for k, x in sf_dict_members(want): shards.setdefault(sm.get(k, "?"), []).append("%s=%s" % (k, x)) if {h: ",".join(ms) for h, ms in shards.items()} != v["shards"]: fails.append("delivery-only %s: header shards" % v["name"]) # --------------------------------------------------------------------------------------- def read_reasons(): with open(EXTENSION, encoding="utf-8") as fh: text = fh.read() at = text.index("\n## 11. Reasons") end = text.index("\n## ", at + 1) return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")] REASONS = read_reasons() REUSED = ["truncated", "malformed", "unknown_version", "bad_signature", "vdaf_verify_error", "hpke_decrypt_error", "report_replayed", "invalidBatchSize", "batchMismatch", "batchOverlap", "batchInvalid"] MEMO = {} def memo(key, fn): """A part's failures, computed once per input: a mutated copy re-runs only what it changed.""" k = hashlib.sha256(json.dumps(key, sort_keys=True).encode()).hexdigest() if k not in MEMO: MEMO[k] = fn() return list(MEMO[k]) def check(doc, vdaf_files, quiet=False): fails = [] for n, d in sorted(vdaf_files.items()): fails += memo(["vdaf", n, d, sorted(SKIP)], lambda: run_xof(d) if n == "XofTurboShake128.json" else run_vdaf_file(n[:-5], d)) if not quiet: print(("PASS" if not fails else "FAIL") + " the draft's Prio3 and XofTurboShake128 vectors (%d files)" % len(vdaf_files)) for f in doc["flows"]: named = [x for x in doc["receipts"] if x["name"] == f["name"]] def run(f=f): out = [] check_flow(doc, f, out, quiet) return out fails += run() if not quiet else memo(["flow", f, doc["keys"], doc["sessions"], named, sorted(SKIP & {"proof", "floor"})], run) for v in doc["receipts"]: got = receipt_outcome(v) if got != v["outcome"]: fails.append("receipt %s: stated %s, got %s" % (v["name"], v["outcome"], got)) for v in doc["cells"]: got = cell_outcome(doc, v) if got != v["outcome"]: fails.append("cell %s: stated %s, got %s" % (v["name"], v["outcome"], got)) if got.get("outcome") == "accept" and (v.get("plaintext") != plaintext_cell(doc["sessions"]) or got["cell"] != v["plaintext"]): fails.append("cell %s: not the plaintext cell of live-metering §4.7" % v["name"]) for v in doc["delivery_only"]: check_delivery_only(v, fails) if doc.get("codes") != REASONS: fails.append("the corpus's codes are not §11's") if not quiet: print(("PASS" if not fails else "FAIL") + " %d receipts, %d cells, %d delivery-only reports" % (len(doc["receipts"]), len(doc["cells"]), len(doc["delivery_only"]))) return fails def reasons_given(doc): out = set() for sec in ("receipts", "cells", "delivery_only"): out |= {v["outcome"].get("reason") for v in doc[sec]} for f in doc["flows"]: out |= {o["outcome"] for o in f["job"]["outcomes"]} | {e["error"] for e in f["upload_errors"]} | {r["outcome"] for r in f["refusals"]} return out def flip_hex(h, i): b = bytearray(bytes.fromhex(h)) b[i] ^= 0x01 return b.hex() def mutations(doc, vdaf_files): def named(d, sec, name): return next(x for x in d[sec] if x["name"] == name) def m(label, fn, vf=None): d = copy.deepcopy(doc) v = copy.deepcopy(vdaf_files) (fn or (lambda _: None))(d) if vf: vf(v) return label, d, v yield m("a report's leader share", lambda d: d["flows"][0].update(upload=flip_hex(d["flows"][0]["upload"], 300))) yield m("a VerifyInit's payload", lambda d: d["flows"][1]["job"].update(init_req=flip_hex(d["flows"][1]["job"]["init_req"], len(d["flows"][1]["job"]["init_req"]) // 2 - 1))) yield m("a VerifyResp", lambda d: d["flows"][2]["job"].update(resp=flip_hex(d["flows"][2]["job"]["resp"], 40))) yield m("a session's stalls", lambda d: d["sessions"][3]["metrics"].update(stalls=d["sessions"][3]["metrics"]["stalls"] + 1)) yield m("a session's start time", lambda d: d["sessions"][5]["metrics"].update(vst_ms=d["sessions"][5]["metrics"]["vst_ms"] + 4000)) yield m("the out-of-range report stated committed", lambda d: d["flows"][3]["job"]["outcomes"][50].update(outcome="committed")) yield m("a listed measurement", lambda d: d["flows"][0]["reports"][7].update(measurement=[1, 0, 0, 0])) yield m("a refusal's outcome", lambda d: d["flows"][0]["refusals"][0].update(outcome="batchMismatch")) yield m("a refusal's batch interval", lambda d: d["flows"][0]["refusals"][0].update(request=flip_hex(d["flows"][0]["refusals"][0]["request"], 18))) yield m("a flow's result", lambda d: d["flows"][2]["result"].__setitem__(1, d["flows"][2]["result"][1] + 1)) yield m("the helper's sealed statement", lambda d: d["flows"][1]["collection"].update(helper_statement=flip_hex(d["flows"][1]["collection"]["helper_statement"], 60))) yield m("the collector's private key", lambda d: d["keys"]["collector_hpke"].update(private_key=flip_hex(d["keys"]["collector_hpke"]["private_key"], 5))) yield m("the verification key seed", lambda d: d["keys"].update(verify_key_seed=flip_hex(d["keys"]["verify_key_seed"], 0))) yield m("an accepted receipt's signature", lambda d: named(d, "receipts", "stalls").update(receipt=flip_hex(named(d, "receipts", "stalls")["receipt"], -3 + len(named(d, "receipts", "stalls")["receipt"]) // 2))) yield m("an accepted receipt's stated count", lambda d: named(d, "receipts", "durations")["outcome"].update(report_count=51)) yield m("an accepted receipt's reports", lambda d: named(d, "receipts", "start_time")["reports"].pop()) yield m("a refused receipt stated accepted", lambda d: named(d, "receipts", "batch_below_the_minimum").update(outcome={"outcome": "accept"})) yield m("a refusal's place", lambda d: named(d, "receipts", "helper_statement_by_another")["outcome"].update(at="leader")) yield m("a cell's plaintext", lambda d: named(d, "cells", "room_lesson")["plaintext"].update(stalls=0)) yield m("a delivery-only report given a measurement key", lambda d: named(d, "delivery_only", "audio_request").update( delivery_only=named(d, "delivery_only", "audio_request")["delivery_only"] + ",sn=4")) yield m("a delivery-only refusal's key", lambda d: named(d, "delivery_only", "cohort_served_as_delivery")["outcome"].update(key="d")) yield m("the codes reordered", lambda d: d["codes"].reverse()) yield m("a draft vector's result", None, lambda v: v["Prio3Sum_2.json"].update(agg_result=1522)) yield m("a draft vector's verifier share", None, lambda v: v["Prio3Histogram_0.json"]["reports"][0]["verifier_shares"][0].__setitem__( 0, flip_hex(v["Prio3Histogram_0.json"]["reports"][0]["verifier_shares"][0][0], 3))) yield m("a draft vector's helper seed", None, lambda v: v["Prio3SumVec_1.json"]["reports"][0]["input_shares"].__setitem__( 1, flip_hex(v["Prio3SumVec_1.json"]["reports"][0]["input_shares"][1], 0))) CHECKS = { "task-rules": "§4.3 rule 1: the task's fields", "measure": "§4.3 rule 3: the measure's VDAF", "floor-rule": "§4.3 rule 4: a minimum batch size of at least fifty", "one-key": "§4.3 rule 5: two aggregator keys", "collector-signer": "§8.1 step 3: the collector's key", "collector-sig": "§8.1 step 3: the receipt's signature", "collection": "§8.1 step 4: the collection job", "two-aggregators": "§8.1 step 5: two signers in two roles", "statement-signer": "§8.1 step 5: each statement's key", "statement-sig": "§8.1 step 5: each statement's signature", "binding": "§8.1 step 6: the task and collection bound", "batch": "§8.1 step 7: the batch within the query", "agreement": "§8.1 step 8: the statements agree", "floor": "§8.1 step 9: the batch at least the minimum", "share": "§8.1 step 10: each aggregate share", "result": "§8.1 step 10: the result unsharded", "reports": "§8.1 step 11: the batch's reports", "cell": "§9.2: one receipt of each measure, one scope, one batch", "proof": "§3.3: an out-of-range report is refused", } def main(corpus_dir): fails = [] st = ED._self_test() print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st)) fails += ["ed25519 self-test"] if st else [] ct = crypto_self_test() print(("PASS" if not ct else "FAIL") + " TurboSHAKE128, X25519 and HPKE self-tests " + ",".join(ct)) fails += ct print(("PASS" if len(REASONS) == 17 else "FAIL") + " §11 read from the extension: %d reasons" % len(REASONS)) fails += [] if len(REASONS) == 17 else ["§11 does not read"] print(("PASS" if len(CLASSES) == 42 else "FAIL") + " delivery-telemetry §3.4 read: %d keys classed" % len(CLASSES)) fails += [] if len(CLASSES) == 42 else ["§3.4 does not read"] with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as fh: doc = json.load(fh) vdir = os.path.join(corpus_dir, "vdaf-22") vdaf_files = {} for n in sorted(os.listdir(vdir)): with open(os.path.join(vdir, n), encoding="utf-8") as fh: vdaf_files[n] = json.load(fh) fails += check(doc, vdaf_files) extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md", "vdaf-22"}) print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and vdaf-22/") fails += ["an extra file: %s" % n for n in extra] missing = [r for r in REASONS + REUSED if r not in reasons_given(doc)] print(("PASS" if not missing else "FAIL") + " every reason of §11, and every code it reuses, appears in the corpus") fails += ["no case gives %s" % r for r in missing] for label, d, v in mutations(doc, vdaf_files): caught = check(d, v, quiet=True) print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else "")) if not caught: fails.append("a mutated copy verifies: %s" % label) for name, what in CHECKS.items(): SKIP.clear() SKIP.add(name) caught = check(doc, {}, quiet=True) SKIP.clear() print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else "")) if not caught: fails.append("the corpus does not catch a verifier without %s" % what) for x in fails: print("FAIL " + x) print(("PASS" if not fails else "FAIL") + " QoE aggregation corpus") return 1 if fails else 0 if __name__ == "__main__": if len(sys.argv) != 2: print("usage: telemetry_aggregate_verify.py ", file=sys.stderr) sys.exit(2) sys.exit(main(sys.argv[1]))