a4b4fbcac5
Written from the extension's text and the drafts it cites, sharing no code with wai-rs: Keccak-p[1600,12], TurboSHAKE128 and the VDAF XOF; Field64 and Field128, the Lagrange-basis FLP and Prio3 (passing the draft's vectors); X25519, HKDF, AES-128-GCM and HPKE (RFC 9180 A.1.1); every flow re-run from its reports with the test keys, byte for byte; receipts and cells in §8.1's order; delivery-telemetry §3.4's classes read from the text. Every mutated copy of the corpus fails and every weakened verifier fails the corpus. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1822 lines
71 KiB
Python
1822 lines
71 KiB
Python
#!/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 <live-conformance/aggregate>
|
|
|
|
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 <live-conformance/aggregate>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|