61320c8641
telemetry-aggregation §6.1's rule that `rejected` counts no report the aggregator aggregated had no test, and impression_aggregate_verify.py did not apply it. The flight flow now sends the helper a report it committed, and the Python flow filters as the Rust does. impression-aggregates §7.3 says what a reader does with a statement of version 1 beside one of version 2, and the corpus adds a receipt and a table of each. The ledger's corpus takes these receipts, so it is regenerated by the commit after this one. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1554 lines
71 KiB
Python
1554 lines
71 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the impression aggregates corpus (extensions/impression-aggregates.md).
|
|
|
|
Written from the extension's text and the texts it cites. It shares no code with wai-rs, which
|
|
generates the corpus, and never runs it. From the second implementations of the texts this
|
|
extension builds on it takes what those texts define, by reference as the extension does:
|
|
|
|
- telemetry_aggregate_verify.py (telemetry-aggregation, DAP, VDAF, HPKE): Prio3Histogram with its
|
|
field, XOF and FLP; X25519, HKDF-SHA256, AES-128-GCM and HPKE open; DAP's messages; a task's
|
|
`task_info`, its id and its §4.3 rule 1; the statement and the aggregate receipt; and §8.1
|
|
steps 1 to 11, step 2 given as §5.3 here;
|
|
- stitched_presentation_verify.py: a pod-input record under stitched-presentation §7.2 steps 1
|
|
to 3 and the claims under §7.3 steps 1 to 10, with each presentation's quartile (§7.4);
|
|
- ad_presentation_verify.py: a token's reading (§7.1), its TokenChallenge (§7.2), the issuer
|
|
key's form and RSASSA-PSS (§7.3);
|
|
- ../signer-conformance/signer_verify.py: key-log revisions, anchors and confirmation
|
|
(jwp-receipts §7.3, §7.4), `aggregate/task` among its pinned-only authorities (§7.5);
|
|
- rung_price_verify.py: a region code and *holds* (rung-price §2);
|
|
- ../claim-conformance/_ed25519.py: Ed25519 as SPEC §7 *Signatures* states it.
|
|
|
|
Everything this extension adds is written here: the catalogue, its columns and cells (§4); the
|
|
task's rules and the task file with its HpkeConfigLists (§5); the contribution, the upload file,
|
|
the upload plan and the settlement (§6); the aggregators' report rule, token check and hold
|
|
(§7); a receipt's §8.1 steps 12 and 13, and the table (§8.2). Each flow is re-run end to end:
|
|
every report's two input shares opened with the aggregators' test keys, its extensions and token
|
|
checked as each aggregator checks its own share, each presentation's contribution recomputed
|
|
and its measurement recombined and required to be one at its cell, both aggregators' Prio3
|
|
verification, each upload's response and each relay's merge, the aggregation job's request and
|
|
response byte for byte, every bucket, collection (held, refused or completed), aggregate share,
|
|
statement, result and receipt.
|
|
|
|
For every case it requires the outcome the corpus states, field for field; that the corpus
|
|
lists exactly §10's codes (read from the extension), that each of them and each code §10 says
|
|
the extension reuses appears in an outcome, and that no outcome gives a code the extension does
|
|
not name; and it checks the size bounds no vector carries. Then it 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 -I impression_aggregate_verify.py <live-conformance/impressions>
|
|
|
|
Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
|
|
is caught. Python 3.8 or later, the standard library and the signer verifier's own requirement
|
|
(the `cryptography` package).
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
for _d in ("claim-conformance", "signer-conformance"):
|
|
sys.path.insert(0, os.path.join(HERE, "..", _d))
|
|
sys.path.insert(0, HERE)
|
|
import _ed25519 as ED # noqa: E402
|
|
import ad_presentation_verify as APV # noqa: E402
|
|
import rung_price_verify as RP # noqa: E402
|
|
import signer_verify as SG # noqa: E402
|
|
import stitched_presentation_verify as SPV # noqa: E402
|
|
import telemetry_aggregate_verify as TAV # noqa: E402
|
|
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "impression-aggregates.md")
|
|
JWP = os.path.join(HERE, "..", "extensions", "jwp-receipts.md")
|
|
|
|
Refuse = TAV.Refuse
|
|
CAT_MAGIC, TASK_MAGIC, UPLOAD_MAGIC = b"WIC1", b"WIT1", b"WIU1"
|
|
TASK_DOMAIN = b"wai:impression-task\x01" # §5.4
|
|
ATTEST_DOMAIN = b"wai:impression-report-attest\x01" # §7.2
|
|
TOKEN_EXT = 0x5749 # §7.2 `wai_impression_token`
|
|
MAX_CATALOGUE, MAX_TASK_FILE, MAX_UPLOAD = 1 << 20, 1 << 20, 1 << 24
|
|
MAX_CELLS = 4096 # §4.3 rule 7
|
|
FLOOR = 50 # §5.3 step 5, §8.2 `floor_met`
|
|
MIN_PRECISION = 3600 # §5.3 step 3
|
|
LAST_S = 253402300799 # §5.3 step 3: the last instant, in seconds
|
|
HEADER = 41 # §6.5: a WIU1's bytes before its reports
|
|
SUITE = (0x0020, 0x0001, 0x0001) # DAP §7: X25519, HKDF-SHA256, AES-128-GCM
|
|
CLAIM, ATTEST = 1, 2 # ad-presentation §8
|
|
REPORT_ERRORS = {1: "batch_collected", 2: "report_replayed", 3: "report_dropped", 4: "hpke_unknown_config_id",
|
|
5: "hpke_decrypt_error", 6: "vdaf_verify_error", 7: "invalid_message", 8: "report_too_early",
|
|
9: "unknown_verification_key_id", 10: "unsupported_extension"} # DAP §4.1
|
|
CODE = {v: k for k, v in REPORT_ERRORS.items()}
|
|
KEEP = 8 # §6.6: report_too_early is kept
|
|
PROBLEMS = ["invalidBatchSize", "batchMismatch", "batchOverlap", "batchInvalid", "invalidAggregationParameter",
|
|
"invalidMessage"] # DAP §3.6, as the collections meet them
|
|
|
|
# A weakened run drops one of these checks (see `WEAKENINGS`); empty checks everything.
|
|
SKIP = set()
|
|
|
|
|
|
def drops(name):
|
|
return name in SKIP
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for x in parts:
|
|
h.update(x)
|
|
return h.digest()
|
|
|
|
|
|
def u32(n):
|
|
return n.to_bytes(4, "big")
|
|
|
|
|
|
def section_codes(path, heading, column):
|
|
"""The codes of a reasons table: the `column`-th cell of each row that starts with a code."""
|
|
with open(path, encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
at = text.index(heading)
|
|
end = text.find("\n## ", at + 1)
|
|
rows = text[at:end if end >= 0 else len(text)].splitlines()
|
|
out = []
|
|
for r in rows:
|
|
cells = [c.strip() for c in r.split("|")]
|
|
if len(cells) > column + 1 and cells[column].startswith("`"):
|
|
out.append(cells[column].strip("`"))
|
|
return out
|
|
|
|
|
|
REASONS = section_codes(EXTENSION, "\n## 10. Reasons", 1)
|
|
JWP_REASONS = section_codes(JWP, "\n### 7.7. Reasons", 3)
|
|
|
|
|
|
# Each of these is a pure function of its inputs, so each is computed once for a set of inputs.
|
|
MEMO = {}
|
|
|
|
|
|
def memo(key, fn):
|
|
if key not in MEMO:
|
|
try:
|
|
MEMO[key] = (True, fn())
|
|
except (ValueError, Refuse) as e:
|
|
MEMO[key] = (False, e)
|
|
ok, v = MEMO[key]
|
|
if not ok:
|
|
raise v
|
|
return v
|
|
|
|
|
|
# ------------------------------------------------------------------ §4 the catalogue
|
|
|
|
def read_terms(r):
|
|
"""stitched-presentation §4.1's measurement terms, little-endian there: (bytes, fields)."""
|
|
start = r.i
|
|
q = {"require": r.u(1), "rule": r.u(1), "token_type": int.from_bytes(r.take(2), "little")}
|
|
q["issuer"] = r.take(int.from_bytes(r.take(2), "little"))
|
|
return r.b[start:r.i], q
|
|
|
|
|
|
def terms_bytes(q):
|
|
return struct.pack("<BBHH", q["require"], q["rule"], q["token_type"], len(q["issuer"])) + q["issuer"]
|
|
|
|
|
|
def terms_ok(q):
|
|
"""§4.3 rule 2: ad-presentation §8's rules, bit 0 (claim) set, and attest by token type 2 only."""
|
|
r, rule, tt, iss = q["require"], q["rule"], q["token_type"], q["issuer"]
|
|
claim = bool(r & CLAIM)
|
|
if r > 7 or (r & 6 and not claim) or rule > 2 or (rule == 0) == claim:
|
|
return False
|
|
if r & ATTEST:
|
|
if tt not in (1, 2) or not iss:
|
|
return False
|
|
elif tt != 0 or iss:
|
|
return False
|
|
return claim and (not r & ATTEST or tt == 2)
|
|
|
|
|
|
def ascending(xs):
|
|
return all(a < b for a, b in zip(xs, xs[1:]))
|
|
|
|
|
|
def ceil_sqrt(n):
|
|
s = math.isqrt(n)
|
|
return s if s * s >= n else s + 1
|
|
|
|
|
|
def rsa_key_ok(k):
|
|
try:
|
|
APV.rsa_key(k)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def read_catalogue(b):
|
|
"""§4.3: the catalogue, or `catalogue_malformed`."""
|
|
bad = Refuse("catalogue_malformed", "catalogue")
|
|
if len(b) > MAX_CATALOGUE:
|
|
raise bad
|
|
r = TAV.Rd(b)
|
|
try:
|
|
magic, version = r.take(4), r.u(1)
|
|
if magic != CAT_MAGIC or (version != 1 and not drops("cat-version")): # 1
|
|
raise bad
|
|
raw, q = read_terms(r)
|
|
hold = r.u(4)
|
|
signers = [r.take(32) for _ in range(r.u(1))]
|
|
creatives = [r.take(32) for _ in range(r.u(1))]
|
|
regions = [r.opaque(1) for _ in range(r.u(1))]
|
|
keys = [r.opaque(2) for _ in range(r.u(1))]
|
|
r.end()
|
|
except ValueError:
|
|
raise bad
|
|
attest = bool(q["require"] & ATTEST)
|
|
cells = len(creatives) * (len(regions) + 1) * 5
|
|
rules = [
|
|
("cat-terms", not terms_ok(q)), # 2
|
|
("cat-hold", hold == 0), # 3
|
|
("cat-signers", not 1 <= len(signers) <= 16), # 4
|
|
("cat-order", not ascending(signers)),
|
|
("cat-creatives", not creatives), # 5
|
|
("cat-order", not ascending(creatives)),
|
|
("cat-region", not all(RP.region_code(x.decode("latin-1")) for x in regions)), # 6
|
|
("cat-order", not ascending(regions)),
|
|
("cat-overlap", any(len(x) == 2 and any(y.startswith(x + b"-") for y in regions) for x in regions)),
|
|
("cat-cells", cells > MAX_CELLS), # 7
|
|
("cat-keys", (not attest and keys) or (attest and not 1 <= len(keys) <= 4)), # 8
|
|
("cat-key-form", not all(rsa_key_ok(k) for k in keys)),
|
|
("cat-order", not ascending(keys)),
|
|
]
|
|
if any(broken and not drops(name) for name, broken in rules):
|
|
raise bad
|
|
return {"bytes": b, "digest": sha(b), "terms_raw": raw, "terms": q, "attest": attest, "hold_s": hold,
|
|
"signers": signers, "creatives": creatives, "regions": regions, "keys": keys, "cells": cells,
|
|
"chunk": ceil_sqrt(cells) if cells else 0}
|
|
|
|
|
|
def column(cat, g):
|
|
"""§4.2: the column of a receiver's region `g` (bytes or None); `r` is `other`."""
|
|
other = len(cat["regions"])
|
|
if g is None:
|
|
return other
|
|
s = g.decode("latin-1")
|
|
if not RP.region_code(s):
|
|
return other
|
|
for j, x in enumerate(cat["regions"]):
|
|
if (x == g) if drops("region-subdivision") else RP.holds([x.decode("ascii")], s):
|
|
return j
|
|
return other
|
|
|
|
|
|
def column_name(cat, j):
|
|
return cat["regions"][j].decode("ascii") if j < len(cat["regions"]) else "other"
|
|
|
|
|
|
def cell_of(cat, i, j, q):
|
|
return (i * (len(cat["regions"]) + 1) + j) * 5 + (q - 1)
|
|
|
|
|
|
def catalogue_outcome(b):
|
|
try:
|
|
c = read_catalogue(b)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason}
|
|
return {"outcome": "read", "digest": c["digest"].hex(), "cells": c["cells"], "chunk": c["chunk"]}
|
|
|
|
|
|
# ------------------------------------------------------------------ §5 the task
|
|
|
|
def task_rules(cfg, cat):
|
|
"""§5.3 against a catalogue that reads: the task's fields, Prio3, id, flight and retention."""
|
|
t, info = TAV.task_fields(cfg) # 1
|
|
if not drops("catalogue-digest") and (info["measure"] != 5 or info["scope"] != cat["digest"]):
|
|
raise Refuse("catalogue_mismatch", "task") # 2
|
|
got = {}
|
|
|
|
def steps_3_4(t, info):
|
|
ex = t["extensions"]
|
|
if t["time_precision"] < MIN_PRECISION and not drops("task-precision"):
|
|
raise Refuse("task_invalid", "task")
|
|
if len(ex) != 1 or ex[0][0] != 1 or len(ex[0][1]) != 16:
|
|
raise Refuse("task_invalid", "task")
|
|
iv = TAV.interval_of(ex[0][1])
|
|
if iv[1] < 1 and not drops("task-interval"):
|
|
raise Refuse("task_invalid", "task")
|
|
retention = (iv[0] + iv[1]) * t["time_precision"] + cat["hold_s"]
|
|
if retention > LAST_S and not drops("task-retention"): # a u64 overflow is past it too
|
|
raise Refuse("task_invalid", "task")
|
|
config = u32(cat["cells"]) + (b"" if drops("chunk-length") else u32(cat["chunk"]))
|
|
given = t["vdaf_config"][:4] if drops("chunk-length") else t["vdaf_config"]
|
|
if t["vdaf_type"] != 4 or given != config: # 4: DAP Appendix B.4
|
|
raise Refuse("measure_mismatch", "task")
|
|
got.update(interval=iv, retention_s=retention)
|
|
return TAV.Prio3(TAV.Histogram(cat["cells"], cat["chunk"]))
|
|
|
|
t, info, p, tid = TAV.check_task(cfg, measure5=steps_3_4) # 3, 4; then 5, 6
|
|
if "interval" not in got:
|
|
raise ValueError("not a task of measure 5")
|
|
return {"cfg": cfg, "t": t, "info": info, "p": p, "tid": tid, "interval": got["interval"],
|
|
"retention_s": got["retention_s"]}
|
|
|
|
|
|
def task_outcome(cfg, cat_bytes):
|
|
try:
|
|
k = task_rules(cfg, read_catalogue(cat_bytes))
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason}
|
|
return {"outcome": "accept", "task_id": k["tid"].hex(), "retention_s": k["retention_s"]}
|
|
|
|
|
|
def suite_config(configs):
|
|
"""§6.3: the first configuration of DAP §7's suite with a 32-byte key."""
|
|
if drops("hpke-first"):
|
|
return configs[0]
|
|
return next((c for c in configs if (c["kem"], c["kdf"], c["aead"]) == SUITE and len(c["pk"]) == 32), None)
|
|
|
|
|
|
def read_hpke_list(r):
|
|
"""DAP §4.4.1's HpkeConfigList<10..2^16-1>, with its own u16 length first (§5.4 step 1)."""
|
|
body = TAV.Rd(r.opaque(2))
|
|
if len(body.b) < 10:
|
|
raise ValueError("short list")
|
|
out = []
|
|
while not body.done():
|
|
c = {"id": body.u(1), "kem": body.u(2), "kdf": body.u(2), "aead": body.u(2), "pk": body.opaque(2)}
|
|
if not c["pk"]:
|
|
raise ValueError("empty key")
|
|
out.append(c)
|
|
ids = [c["id"] for c in out]
|
|
if len(set(ids)) != len(ids) and not drops("tf-hpke-ids"):
|
|
raise ValueError("two configurations of one id")
|
|
suite = [c for c in out if (c["kem"], c["kdf"], c["aead"]) == SUITE and len(c["pk"]) == 32]
|
|
if not suite and not drops("tf-hpke-suite"):
|
|
raise ValueError("no configuration of DAP's suite")
|
|
return out
|
|
|
|
|
|
def read_task_file(b):
|
|
"""§5.4 step 1: the file, or `task_file_malformed`."""
|
|
bad = Refuse("task_file_malformed", "task_file")
|
|
if len(b) > MAX_TASK_FILE and not drops("tf-bound"):
|
|
raise bad
|
|
r = TAV.Rd(b)
|
|
try:
|
|
magic, version = r.take(4), r.u(1)
|
|
if magic != TASK_MAGIC or (version != 1 and not drops("tf-version")):
|
|
raise bad
|
|
tf = {"task": r.opaque(2), "catalogue": r.opaque(4), "leader_hpke": read_hpke_list(r),
|
|
"helper_hpke": read_hpke_list(r), "signer": r.take(32)}
|
|
sid = r.opaque(2)
|
|
tf["body"] = b[:r.i]
|
|
tf["sig"] = r.take(64)
|
|
r.end()
|
|
except ValueError:
|
|
raise bad
|
|
if not sid or (any(c > 0x7F for c in sid) and not drops("tf-signer-ascii")):
|
|
raise bad
|
|
tf["signer_id"] = sid.decode("latin-1")
|
|
tf["digest"] = sha(tf["body"])
|
|
return tf
|
|
|
|
|
|
def confirmation(tf, revision_texts, anchors):
|
|
"""§5.4 step 5: jwp-receipts §7.4 for `aggregate/task`, the item untimed."""
|
|
held = SG.held_revisions(revision_texts)
|
|
if drops("tf-revocation"): # a receiver that takes in no revision the bearer brings
|
|
named = {a["revision"] for a in anchors}
|
|
held = [x for x in held if SG.rev_hash(x) in named]
|
|
if drops("tf-pinned"):
|
|
anchors = [dict(a, path="pinned", pinned_revision=a["revision"]) for a in anchors]
|
|
authorities = [] if drops("tf-authority") else ["aggregate/task"]
|
|
return SG.confirm(tf["signer"], tf["signer_id"], authorities, None, held, anchors, None)
|
|
|
|
|
|
def accept_task_file(b, revision_texts, anchors):
|
|
"""§5.4, in order: (file, catalogue, task, confirmation), or the refusal."""
|
|
tf = read_task_file(b) # 1
|
|
cat = read_catalogue(tf["catalogue"]) # 2
|
|
task = task_rules(tf["task"], cat) # 3
|
|
if not drops("tf-signature"): # 4
|
|
why = ED.verify(tf["signer"], TASK_DOMAIN + tf["body"], tf["sig"])
|
|
if why:
|
|
raise Refuse(why, "task_file")
|
|
v = confirmation(tf, revision_texts, anchors) # 5
|
|
if not v.startswith("confirmed"):
|
|
raise Refuse(re.search(r"reason=(\S+)", v).group(1), "task_file")
|
|
return tf, cat, task, v
|
|
|
|
|
|
def task_file_outcome(doc, c):
|
|
fx = doc["fixtures"]
|
|
try:
|
|
tf, _, task, v = accept_task_file(bytes.fromhex(c["bytes"]), [fx["key_logs"][n] for n in c["revisions"]],
|
|
[fx["anchors"][n] for n in c["anchors"]])
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason}
|
|
return {"outcome": "accept", "digest": tf["digest"].hex(), "task_id": task["tid"].hex(), "confirmation": v}
|
|
|
|
|
|
def receiver_task_file(doc, name):
|
|
"""A task file of the fixtures, accepted under what `receiver` holds (§6.1)."""
|
|
fx = doc["fixtures"]
|
|
rv = fx["receiver"]
|
|
return accept_task_file(bytes.fromhex(fx["task_files"][name]), [fx["key_logs"][n] for n in rv["revisions"]],
|
|
[fx["anchors"][n] for n in rv["anchors"]])
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.2 the token
|
|
|
|
def pss(key, msg, sig):
|
|
return memo(("pss", key, msg, sig), lambda: APV.pss_verify(key, msg, sig))
|
|
|
|
|
|
def token_check(cat, tid, rid, token):
|
|
"""§7.2's token: None when it is type 0x0002, answers the challenge of `tid` and `rid`, names one
|
|
of the catalogue's keys and verifies under it; otherwise `attestation_invalid`."""
|
|
t = APV.read_token(token)
|
|
if t is None or t["type"] != 2:
|
|
return "attestation_invalid"
|
|
ctx = sha(ATTEST_DOMAIN, tid, rid)
|
|
if not drops("token-challenge") and t["digest"] != sha(APV.token_challenge(2, cat["terms"]["issuer"], ctx)):
|
|
return "attestation_invalid"
|
|
key = next((k for k in cat["keys"] if sha(k) == t["key_id"]), None)
|
|
if key is None or not pss(key, t["input"], t["auth"]):
|
|
return "attestation_invalid"
|
|
return None
|
|
|
|
|
|
def report_rule(cat, tid, rid, public, private, role):
|
|
"""§7.2 at DAP §4.5.3.4, an aggregator's own input share: None, or (error, WAI reason)."""
|
|
if public and not drops("public-extension"):
|
|
return "invalid_message", None
|
|
if not cat["attest"]:
|
|
return ("invalid_message", None) if private else None
|
|
if role == 3 and drops("helper-token"):
|
|
return None
|
|
if len(private) != 1 or private[0][0] != TOKEN_EXT:
|
|
return "invalid_message", "attestation_absent"
|
|
if token_check(cat, tid, rid, private[0][1]):
|
|
return "invalid_message", "attestation_invalid"
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------------ §6.2 the contribution
|
|
|
|
def held_fixtures(doc):
|
|
fx = doc["fixtures"]
|
|
manifests = [fx["manifests"][n].encode("utf-8") for n in sorted(fx["manifests"])]
|
|
creatives = [[{"content_hash": bytes.fromhex(o["content_hash"]), "duration_ms": o["duration_ms"]}
|
|
for o in fx["creatives"][n]["objects"]] for n in sorted(fx["creatives"])]
|
|
return manifests, creatives
|
|
|
|
|
|
def judge(doc, record, claims, signers):
|
|
"""stitched-presentation §7.2 steps 1 to 3 for the record (no other record, no response held)
|
|
and §7.3 steps 1 to 10 for the claims, in order."""
|
|
manifests, creatives = held_fixtures(doc)
|
|
key = ("judge", record, tuple(claims), tuple(signers), tuple(manifests),
|
|
json.dumps(doc["fixtures"]["creatives"], sort_keys=True))
|
|
inp = {"records": [record], "claims": claims, "manifests": manifests, "responses": [], "receipts": [],
|
|
"creatives": creatives, "issuers": [], "signers": signers}
|
|
return memo(key, lambda: SPV.verify(inp, through_attestation=False))
|
|
|
|
|
|
def no_report(reason, discarded=()):
|
|
return {"outcome": "no_report", "reason": reason, "discarded": list(discarded)}
|
|
|
|
|
|
def contribute(doc, cat, task, record, claims, region, at_ms, pool):
|
|
"""§6.2, in order."""
|
|
v = judge(doc, record, claims, cat["signers"])
|
|
verdict = v["record_facts"][0]["verdict"]
|
|
if verdict != "accepted": # 1
|
|
return no_report(verdict)
|
|
rec = SPV.read_record(record)
|
|
if rec["session"] != bytes(16) and not drops("shared"): # 2
|
|
return no_report("record_not_shared")
|
|
if terms_bytes(rec["terms"]) != cat["terms_raw"] and not drops("terms-equal"): # 3
|
|
return no_report("record_terms_differ")
|
|
cf = v["claim_facts"][-1]
|
|
if cf["verdict"] != "presented": # 4
|
|
return no_report(cf["verdict"])
|
|
ad = rec["ads"][cf["slot"]]
|
|
if ad["creative_hash"] not in cat["creatives"] and not drops("creative-listed"): # 5
|
|
return no_report("creative_not_listed")
|
|
tp = task["t"]["time_precision"]
|
|
if at_ms is None and not drops("instant"): # 6
|
|
return no_report("instant_unknown")
|
|
time = task["interval"][0] if at_ms is None else at_ms // (1000 * tp)
|
|
iv = task["interval"]
|
|
if not iv[0] <= time < iv[0] + iv[1] and not drops("flight-window"): # 7
|
|
return no_report("outside_task")
|
|
discarded, rid = [], "fresh"
|
|
if cat["attest"]: # 8
|
|
chosen = None
|
|
for pair in sorted(pool, key=lambda x: bytes.fromhex(x["id"])):
|
|
if drops("token-pool") or token_check(cat, task["tid"], bytes.fromhex(pair["id"]),
|
|
bytes.fromhex(pair["token"])) is None:
|
|
chosen = pair
|
|
break
|
|
discarded.append(pair["id"])
|
|
if chosen is None:
|
|
return no_report("token_exhausted", discarded)
|
|
rid = chosen["id"]
|
|
q = cf["quartile"]
|
|
j = column(cat, region)
|
|
i = cat["creatives"].index(ad["creative_hash"]) if ad["creative_hash"] in cat["creatives"] else None
|
|
return {"outcome": "report", "cell": None if i is None else cell_of(cat, i, j, q), "creative": i,
|
|
"column": column_name(cat, j), "quartile": q, "time": time, "id": rid, "discarded": discarded}
|
|
|
|
|
|
def contribution_outcome(doc, c):
|
|
tf, cat, task, _ = receiver_task_file(doc, c["task_file"])
|
|
record = bytes.fromhex(doc["fixtures"]["records"][c["record"]]["bytes"])
|
|
region = None if c["region"] is None else c["region"].encode("utf-8")
|
|
return contribute(doc, cat, task, record, [bytes.fromhex(x) for x in c["claims"]], region, c["at_ms"], c["pool"])
|
|
|
|
|
|
# ------------------------------------------------------------------ §6.4 to §6.6
|
|
|
|
def read_upload(b):
|
|
"""§6.4: (task_id, reports), each report with its bytes; or `upload_malformed`."""
|
|
bad = Refuse("upload_malformed", "upload")
|
|
if len(b) > MAX_UPLOAD and not drops("upload-bound"):
|
|
raise bad
|
|
r = TAV.Rd(b)
|
|
try:
|
|
magic, version = r.take(4), r.u(1)
|
|
if magic != UPLOAD_MAGIC or (version != 1 and not drops("upload-version")):
|
|
raise bad
|
|
tid = r.take(32)
|
|
n = r.u(4)
|
|
up = TAV.Rd(r.take(min(n, len(b) - r.i) if drops("upload-length") else n))
|
|
r.end()
|
|
reports = []
|
|
while not up.done():
|
|
start = up.i
|
|
rep = TAV.read_report(up)
|
|
rep["raw"] = up.b[start:up.i]
|
|
reports.append(rep)
|
|
except ValueError:
|
|
raise bad
|
|
ids = [x["m"]["id"] for x in reports]
|
|
if not reports or (not ascending(ids) and not drops("upload-order")):
|
|
raise bad
|
|
return tid, reports
|
|
|
|
|
|
def write_upload(tid, raws):
|
|
body = b"".join(raws)
|
|
return UPLOAD_MAGIC + b"\x01" + tid + u32(len(body)) + body
|
|
|
|
|
|
def upload_outcome(b):
|
|
try:
|
|
tid, reports = read_upload(b)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason}
|
|
return {"outcome": "read", "task_id": tid.hex(), "ids": [x["m"]["id"].hex() for x in reports]}
|
|
|
|
|
|
def read_one_report(b):
|
|
r = TAV.Rd(b)
|
|
rep = TAV.read_report(r)
|
|
r.end()
|
|
rep["raw"] = b
|
|
return rep
|
|
|
|
|
|
def plan(task, cat, stored, t_ms, budget):
|
|
"""§6.5."""
|
|
tp, iv = task["t"]["time_precision"], task["interval"]
|
|
timely, late, drop = [], [], []
|
|
for b in stored:
|
|
rep = read_one_report(b)
|
|
rid, tau = rep["m"]["id"], rep["m"]["time"]
|
|
if not iv[0] <= tau < iv[0] + iv[1] or (t_ms >= task["retention_s"] * 1000 and not drops("plan-retention")):
|
|
drop.append(rid)
|
|
continue
|
|
deadline = ((tau + 1) * tp + cat["hold_s"]) * 1000
|
|
(timely if t_ms < deadline or drops("plan-deadline") else late).append((deadline, rid, b))
|
|
send, defer, total, taking = [], [], HEADER, True
|
|
for cls, group in (("timely", sorted(timely)), ("late", sorted(late))):
|
|
for _, rid, b in group:
|
|
if taking and (budget is None or total + len(b) <= budget or drops("plan-budget")):
|
|
send.append((rid, cls, b))
|
|
total += len(b)
|
|
else:
|
|
taking = False
|
|
defer.append(rid)
|
|
raws = [b for _, _, b in sorted(send)]
|
|
return {"send": [{"id": rid.hex(), "class": cls} for rid, cls, _ in send], "defer": [x.hex() for x in defer],
|
|
"drop": [x.hex() for x in sorted(drop)], "file": write_upload(task["tid"], raws).hex() if send else None}
|
|
|
|
|
|
def plan_outcome(doc, c):
|
|
_, cat, task, _ = receiver_task_file(doc, c["task_file"])
|
|
return plan(task, cat, [bytes.fromhex(x) for x in c["stored"]], c["t_ms"], c["budget"])
|
|
|
|
|
|
DELETE = {1, 2, 3, 4, 5, 6, 7, 9, 10} # §6.6: DAP §4.1's errors but report_too_early
|
|
|
|
|
|
def settle(sent, response):
|
|
"""§6.6."""
|
|
_, reports = read_upload(sent)
|
|
ids = [x["m"]["id"] for x in reports]
|
|
entries = None
|
|
if len(response) % 17 == 0:
|
|
es = [(response[k:k + 16], response[k + 16]) for k in range(0, len(response), 17)]
|
|
at, ok = 0, True
|
|
for rid, _ in es:
|
|
if drops("settle-subsequence"):
|
|
continue
|
|
while at < len(ids) and ids[at] != rid:
|
|
at += 1
|
|
if at == len(ids):
|
|
ok = False
|
|
break
|
|
at += 1
|
|
entries = es if ok else None
|
|
if entries is None:
|
|
return {"settles": False, "delete": [], "keep": [x.hex() for x in ids]}
|
|
errors = dict(entries)
|
|
delete, keep = [], []
|
|
for rid in ids:
|
|
e = errors.get(rid)
|
|
if e is None or e in DELETE or (e == KEEP and drops("settle-too-early")) or \
|
|
(e not in REPORT_ERRORS and drops("settle-unknown-value")):
|
|
delete.append(rid.hex())
|
|
else:
|
|
keep.append(rid.hex())
|
|
return {"settles": True, "delete": delete, "keep": keep}
|
|
|
|
|
|
# ------------------------------------------------------------------ §8 verification
|
|
|
|
def verify_receipt(receipt, cat_bytes, reports):
|
|
"""§8.1, in order: the outcome as accepted, or the refusal."""
|
|
cat = read_catalogue(cat_bytes)
|
|
task = {}
|
|
|
|
def step2(cfg):
|
|
k = task_rules(cfg, cat)
|
|
task.update(k)
|
|
return k["t"], k["info"], k["p"], k["tid"]
|
|
|
|
out = TAV.verify_receipt(receipt, reports, task_check=step2) # 1 to 11
|
|
batch = (out["batch"]["start"], out["batch"]["duration"])
|
|
if not TAV.within(batch, task["interval"]) and not drops("batch-in-task"): # 12
|
|
raise Refuse("batch_invalid", "leader")
|
|
res, n = out["result"], out["report_count"] # 13
|
|
total = sum(res) % TAV.Field128.P if drops("count-integer") else sum(res)
|
|
elements = drops("count-integer") or all(x <= n for x in res)
|
|
if not elements or (total != n and not drops("count-sum")):
|
|
raise Refuse("count_mismatch", "result")
|
|
out["reports_checked"] = reports is not None
|
|
rc = TAV.read_receipt(receipt) # §8.2: each aggregator's own
|
|
out["rejected"] = {"leader": TAV.read_statement(rc["leader"])["rejected"],
|
|
"helper": TAV.read_statement(rc["helper"])["rejected"]}
|
|
return out, cat, task
|
|
|
|
|
|
def receipt_outcome(c):
|
|
try:
|
|
out, _, _ = verify_receipt(bytes.fromhex(c["receipt"]), bytes.fromhex(c["catalogue"]), c["reports"])
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
return out
|
|
|
|
|
|
def rows(cat, vec):
|
|
out = []
|
|
for i in range(len(cat["creatives"])):
|
|
for j in range(len(cat["regions"]) + 1):
|
|
c = vec[cell_of(cat, i, j, 1):cell_of(cat, i, j, 1) + 5]
|
|
out.append({"creative": i, "column": column_name(cat, j), "presentations": sum(c),
|
|
"quartiles": [sum(c[k:]) for k in range(5)], "floor_met": sum(c) >= FLOOR})
|
|
return out
|
|
|
|
|
|
def table(cat_bytes, task_bytes, given):
|
|
"""§8.2."""
|
|
try:
|
|
cat = read_catalogue(cat_bytes) # 1
|
|
task = task_rules(task_bytes, cat)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason}
|
|
tp = task["t"]["time_precision"]
|
|
entries = []
|
|
for g in given: # 2
|
|
rb = bytes.fromhex(g["receipt"])
|
|
e = {"digest": None}
|
|
try:
|
|
rc = TAV.read_receipt(rb)
|
|
e["digest"] = sha(rc["body"]).hex()
|
|
out, _, _ = verify_receipt(rb, cat_bytes, g["reports"])
|
|
if rc["task"] != task_bytes and not drops("task-equal"):
|
|
raise Refuse("task_mismatch", "task")
|
|
e["out"] = out
|
|
except Refuse as x:
|
|
e["verdict"] = "refuse reason=%s at=%s" % (x.reason, x.at)
|
|
entries.append(e)
|
|
one = {} # 3
|
|
for k, e in enumerate(entries):
|
|
if "out" in e:
|
|
one.setdefault(k if drops("dedupe") else e["digest"], e["out"]["batch"])
|
|
over = set()
|
|
for a, x in one.items():
|
|
for b, y in one.items():
|
|
if a != b and x["start"] < y["start"] + y["duration"] and y["start"] < x["start"] + x["duration"]:
|
|
over.add(a)
|
|
if drops("overlap"):
|
|
over = set()
|
|
stated, standing, totals = [], [], None
|
|
for k, e in enumerate(entries):
|
|
if "out" not in e:
|
|
stated.append({"digest": e["digest"], "verdict": e["verdict"]})
|
|
continue
|
|
key = k if drops("dedupe") else e["digest"]
|
|
if key in over:
|
|
stated.append({"digest": e["digest"], "verdict": "set-aside reason=receipts_overlap"})
|
|
continue
|
|
out = e["out"]
|
|
b = out["batch"]
|
|
checked = out["reports_checked"] or drops("reports-checked")
|
|
end = b["start"] + b["duration"]
|
|
stated.append({"digest": e["digest"], "verdict": "stands",
|
|
"batch": {"start": b["start"], "duration": b["duration"], "from_ms": b["start"] * tp * 1000,
|
|
"to_ms": end * tp * 1000},
|
|
"collectable_from_ms": (end * tp + cat["hold_s"]) * 1000, "report_count": out["report_count"],
|
|
"rejected": out["rejected"], "reports_checked": checked, "rows": rows(cat, out["result"])})
|
|
standing.append((key, out, checked)) # 4
|
|
# Each receipt counted once, its rows summed as exact integers; one given more than once is
|
|
# checked when any of its occurrences was.
|
|
counted, batches, vec = set(), [], [0] * cat["cells"]
|
|
rejected = {"leader": 0 if standing else None, "helper": 0 if standing else None}
|
|
for key, out, _ in standing:
|
|
if key in counted:
|
|
continue
|
|
counted.add(key)
|
|
batches.append({"start": out["batch"]["start"], "duration": out["batch"]["duration"]})
|
|
vec = [a + b for a, b in zip(vec, out["result"])]
|
|
for who in rejected: # §8.2: none when one states none
|
|
n = out["rejected"][who]
|
|
if drops("rejected-none"):
|
|
rejected[who] = (rejected[who] or 0) + (n or 0)
|
|
else:
|
|
rejected[who] = None if rejected[who] is None or n is None else rejected[who] + n
|
|
checked = {key for key, _, c in standing if c}
|
|
every = all(c for _, _, c in standing) if drops("checked-any") else all(k in checked for k in counted)
|
|
totals = {"batches": batches, "reports_checked": every, "rejected": rejected, "rows": rows(cat, vec) if standing else []}
|
|
return {"outcome": "table", "receipts": stated, "totals": totals}
|
|
|
|
|
|
# ------------------------------------------------------------------ §6.3 and §7: a flow, re-run
|
|
|
|
def open_input_share(sk, role, aad, ct):
|
|
"""DAP §4.4.2.1: the PlaintextInputShare's private extensions and payload."""
|
|
def go():
|
|
pt = TAV.hpke_open(ct["enc"], sk, b"dap-20 input share" + bytes([1, role]), aad, ct["payload"])
|
|
r = TAV.Rd(pt)
|
|
exts = TAV.read_exts(r)
|
|
payload = r.opaque(4)
|
|
r.end()
|
|
if not payload:
|
|
raise ValueError("empty payload")
|
|
return exts, payload
|
|
return memo(("open", sk, role, aad, ct["config_id"], ct["enc"], ct["payload"]), go)
|
|
|
|
|
|
def verify_init(p, vk, ctx, agg, nonce, public, share):
|
|
return memo(("init", p.v.length, p.v.chunk, vk, ctx, agg, nonce, public, share),
|
|
lambda: p.verify_init(vk, ctx, agg, nonce, public, share))
|
|
|
|
|
|
def meas_share(p, ctx, agg, share):
|
|
return memo(("meas", p.v.length, p.v.chunk, ctx, agg, share), lambda: p.meas_share(ctx, agg, share))
|
|
|
|
|
|
def in_interval(x, iv):
|
|
return iv[0] <= x < iv[0] + iv[1]
|
|
|
|
|
|
def run_flow(doc, f):
|
|
"""One flow of §11, re-run from its reports: the failures."""
|
|
fails = []
|
|
K = doc["keys"]
|
|
|
|
def bad(what):
|
|
fails.append("flow %s: %s" % (f["name"], what))
|
|
|
|
try:
|
|
tf, cat, task, _ = receiver_task_file(doc, f["task_file"])
|
|
except Refuse as e:
|
|
bad("its task file is refused %s" % e.reason)
|
|
return fails
|
|
t, p, tid, iv, cfg = task["t"], task["p"], task["tid"], task["interval"], task["cfg"]
|
|
tp, hold, floor = t["time_precision"], cat["hold_s"], t["min_batch_size"]
|
|
if tid.hex() != f["task_id"]:
|
|
bad("task id")
|
|
vk = TAV.hkdf_expand(TAV.hkdf_extract(b"verify_key", bytes.fromhex(K["verify_key_seed"])), tid, 32)
|
|
if vk.hex() != f["verify_key"]:
|
|
bad("verification key (telemetry-aggregation §4.4)")
|
|
ctx = b"dap-20" + tid
|
|
lsk, hsk, csk = (bytes.fromhex(K[k]["private_key"]) for k in ("leader_hpke", "helper_hpke", "collector_hpke"))
|
|
lconf, hconf = suite_config(tf["leader_hpke"]), suite_config(tf["helper_hpke"])
|
|
if lconf["pk"] != TAV.x25519(lsk, TAV.BASE) or hconf["pk"] != TAV.x25519(hsk, TAV.BASE):
|
|
bad("the configuration the reports are sealed to is not the aggregator's key (§6.3)")
|
|
leader_ids = [c["id"] for c in tf["leader_hpke"]]
|
|
listed = {}
|
|
for x in f["reports"] + f.get("late", []):
|
|
listed[x["id"]] = x
|
|
late = {x["id"] for x in f.get("late", [])}
|
|
|
|
seen, collected, pending, first, inits = set(), [], [], {}, {}
|
|
rejected_by = {"leader": {}, "helper": {}} # by time: the ids each rejected
|
|
fate, buckets, outcomes_stated = {}, {}, None
|
|
|
|
def merged(q):
|
|
out = {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "times": [], "ids": []}
|
|
for tm in range(q[0], q[0] + q[1]):
|
|
if tm in buckets:
|
|
b = buckets[tm]
|
|
out["l"], out["h"] = TAV.vadd(p.f, out["l"], b["l"]), TAV.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
|
|
|
|
def collected_at(tm):
|
|
return any(in_interval(tm, q) for q in collected)
|
|
|
|
for s in f["steps"]:
|
|
if s["step"] == "upload": # §6.4; DAP §4.4.2.2
|
|
try:
|
|
utid, reps = read_upload(bytes.fromhex(s["file"]))
|
|
except Refuse:
|
|
bad("upload %s does not read" % s["name"])
|
|
continue
|
|
if utid != tid:
|
|
bad("upload %s names another task" % s["name"])
|
|
if "sources" in s:
|
|
byid = {}
|
|
for src in s["sources"]:
|
|
try:
|
|
stid, srep = read_upload(bytes.fromhex(src))
|
|
except Refuse:
|
|
bad("a source of %s does not read" % s["name"])
|
|
continue
|
|
if stid != tid:
|
|
bad("a source of %s names another task" % s["name"])
|
|
for x in srep:
|
|
if byid.setdefault(x["m"]["id"], x["raw"]) != x["raw"]:
|
|
bad("two reports of one id with other bytes merged into one file")
|
|
if write_upload(tid, [byid[k] for k in sorted(byid)]).hex() != s["file"]:
|
|
bad("upload %s is not its sources merged, ascending by id (§6.4)" % s["name"])
|
|
resp, errs = b"", []
|
|
for rep in reps:
|
|
m, rid = rep["m"], rep["m"]["id"]
|
|
first.setdefault(rid.hex(), rep)
|
|
if rep["leader"]["config_id"] not in leader_ids:
|
|
e = "hpke_unknown_config_id"
|
|
elif not in_interval(m["time"], iv) and not drops("upload-dropped"):
|
|
e = "report_dropped"
|
|
elif (rid in seen or collected_at(m["time"])) and not drops("upload-replay"):
|
|
e = "report_replayed"
|
|
else:
|
|
e = None
|
|
seen.add(rid)
|
|
pending.append(rep)
|
|
if e:
|
|
resp += rid + bytes([CODE[e]])
|
|
errs.append({"id": rid.hex(), "error": e})
|
|
if rid not in seen:
|
|
fate[rid.hex()] = (e, None)
|
|
if resp.hex() != s["response"] or errs != s["errors"]:
|
|
bad("upload %s: the leader's UploadErrors" % s["name"])
|
|
if "shares" in s:
|
|
srcs = [[x["m"]["id"] for x in read_upload(bytes.fromhex(src))[1]] for src in s["sources"]]
|
|
got = [b"".join(resp[k:k + 17] for k in range(0, len(resp), 17) if resp[k:k + 16] in ids).hex()
|
|
for ids in srcs]
|
|
if got != s["shares"]:
|
|
bad("upload %s: each source's share of the response (§6.4)" % s["name"])
|
|
elif s["step"] == "aggregate": # DAP §4.5; §7.2
|
|
init = bytes([0]) + u32(0) + (0).to_bytes(2, "big")
|
|
resp, refused, job = b"", [], []
|
|
for rep in pending:
|
|
m, rid = rep["m"], rep["m"]["id"]
|
|
aad = TAV.input_share_aad(tid, cfg, m, rep["public"])
|
|
try:
|
|
lexts, lpay = open_input_share(lsk, 2, aad, rep["leader"])
|
|
except ValueError:
|
|
refused.append((rid, "hpke_decrypt_error", None))
|
|
continue
|
|
if not in_interval(m["time"], iv):
|
|
refused.append((rid, "report_dropped", None))
|
|
continue
|
|
why = report_rule(cat, tid, rid, m["exts"], lexts, 2) if not drops("token-check") else None
|
|
if why:
|
|
refused.append((rid,) + why)
|
|
continue
|
|
try:
|
|
lstate, lvs = verify_init(p, vk, ctx, 0, rid, rep["public"], lpay)
|
|
except ValueError:
|
|
refused.append((rid, "vdaf_verify_error", None))
|
|
continue
|
|
out = b"\x00" + u32(len(lvs)) + lvs
|
|
inits[rid] = m["raw"] + u32(len(rep["public"])) + rep["public"] + TAV.write_ct(rep["helper"]) + \
|
|
u32(len(out)) + out
|
|
init += inits[rid]
|
|
try:
|
|
hexts, hpay = open_input_share(hsk, 3, aad, rep["helper"])
|
|
except ValueError:
|
|
resp += rid + b"\x02" + bytes([CODE["hpke_decrypt_error"]])
|
|
job.append((rid, "hpke_decrypt_error", None))
|
|
continue
|
|
why = report_rule(cat, tid, rid, m["exts"], hexts, 3) if not drops("token-check") else None
|
|
if why:
|
|
resp += rid + b"\x02" + bytes([CODE[why[0]]])
|
|
job.append((rid,) + why)
|
|
continue
|
|
try:
|
|
hstate, hvs = verify_init(p, vk, ctx, 1, rid, rep["public"], hpay)
|
|
msg = p.message(ctx, [lvs, hvs])
|
|
hout = p.verify_next(hstate, msg)
|
|
except ValueError:
|
|
resp += rid + b"\x02" + bytes([CODE["vdaf_verify_error"]])
|
|
job.append((rid, "vdaf_verify_error", None))
|
|
continue
|
|
if collected_at(m["time"]): # DAP §4.5.4.3
|
|
resp += rid + b"\x02" + bytes([CODE["batch_collected"]])
|
|
job.append((rid, "batch_collected", None))
|
|
continue
|
|
fin = b"\x02" + u32(len(msg)) + msg
|
|
resp += rid + b"\x00" + u32(len(fin)) + fin
|
|
lout = p.verify_next(lstate, msg)
|
|
b = buckets.setdefault(m["time"], {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0,
|
|
"ck": bytes(32), "ids": []})
|
|
b["l"] = TAV.vadd(p.f, b["l"], TAV.decode_vec(p.f, lout))
|
|
b["h"] = TAV.vadd(p.f, b["h"], TAV.decode_vec(p.f, hout))
|
|
b["n"] += 1
|
|
b["ck"] = bytes(a ^ c for a, c in zip(b["ck"], sha(rid)))
|
|
b["ids"].append((rid, m["time"]))
|
|
job.append((rid, "committed", None))
|
|
pending = []
|
|
times = {rep["m"]["id"]: rep["m"]["time"] for rep in first.values()}
|
|
for rid, o, why in refused + job:
|
|
fate[rid.hex()] = o if o == "committed" else (o, why)
|
|
if o != "committed": # telemetry-aggregation §6.1
|
|
rejected_by["leader"].setdefault(times[rid], set()).add(rid)
|
|
for rid, o, _ in (refused + job if drops("rejected-own") else job):
|
|
if o != "committed": # the helper never sees what the leader refused
|
|
rejected_by["helper"].setdefault(times[rid], set()).add(rid)
|
|
outcomes = [{"id": rid.hex(), "outcome": o, "reason": why} for rid, o, why in refused + job]
|
|
if init.hex() != s["init_req"]:
|
|
bad("the AggregationJobInitReq is not DAP §4.5.3.1's")
|
|
if resp.hex() != s["resp"]:
|
|
bad("the AggregationJobResp is not DAP §4.5.3.2's")
|
|
if outcomes != s["outcomes"]:
|
|
bad("the reports' outcomes at aggregation")
|
|
outcomes_stated = s["outcomes"]
|
|
elif s["step"] == "replay": # DAP §4.5.4.3: sent to the helper again
|
|
rid = bytes.fromhex(s["id"])
|
|
if rid not in inits or not any(rid == i for b in buckets.values() for i, _ in b["ids"]):
|
|
bad("the report sent again is no report the helper committed")
|
|
continue
|
|
if (bytes([0]) + u32(0) + (0).to_bytes(2, "big") + inits[rid]).hex() != s["init_req"]:
|
|
bad("the AggregationJobInitReq of the report sent again")
|
|
if (rid + b"\x02" + bytes([CODE["report_replayed"]])).hex() != s["resp"]:
|
|
bad("the helper's answer to the report sent again")
|
|
rejected_by["helper"].setdefault(first[rid.hex()]["m"]["time"], set()).add(rid)
|
|
elif s["step"] == "collect": # DAP §4.6; §7.1
|
|
req = bytes.fromhex(s["request"])
|
|
try:
|
|
c = TAV.read_collection(req)
|
|
q = TAV.interval_of(c["config"])
|
|
except ValueError:
|
|
bad("collection %s does not read" % s["name"])
|
|
continue
|
|
if s["batch"] != {"start": q[0], "duration": q[1]}:
|
|
bad("collection %s: its batch" % s["name"])
|
|
mg = merged(q)
|
|
due = (q[0] + q[1]) * tp + hold
|
|
if s["now_s"] < due and not drops("hold"):
|
|
got = "processing"
|
|
elif c["agg_param"]:
|
|
got = "invalidAggregationParameter"
|
|
elif q[1] == 0:
|
|
got = "batchInvalid"
|
|
elif any(q[0] < x[0] + x[1] and x[0] < q[0] + q[1] for x in collected):
|
|
got = "batchOverlap"
|
|
elif mg["n"] < floor:
|
|
got = "invalidBatchSize"
|
|
else:
|
|
got = "completed"
|
|
if got != s["outcome"]:
|
|
bad("collection %s: stated %s, the leader gives %s" % (s["name"], s["outcome"], got))
|
|
if "helper_request" in s: # DAP §4.6.4
|
|
try:
|
|
r = TAV.Rd(bytes.fromhex(s["helper_request"]))
|
|
hc = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4),
|
|
"extensions": TAV.read_exts(r)}
|
|
mode, sel, n, ck = r.u(1), r.opaque(2), r.u(8), r.take(32)
|
|
r.end()
|
|
hq = TAV.interval_of(sel)
|
|
ok = mode == 1 and hc["batch_mode"] == 1 and TAV.interval_of(hc["config"]) == q and hq == q and \
|
|
(n, ck) == (mg["n"], mg["ck"])
|
|
if mg["n"] < floor:
|
|
hgot = "invalidBatchSize"
|
|
else:
|
|
hgot = "batchMismatch" if (n, ck) != (mg["n"], mg["ck"]) else "accept"
|
|
except ValueError:
|
|
ok, hgot = False, "invalidMessage"
|
|
if not ok or hgot != s["helper_outcome"]:
|
|
bad("collection %s: the helper's request or its refusal" % s["name"])
|
|
if got == "completed" and s["outcome"] == "completed":
|
|
collected.append(q)
|
|
aggregated = set() if drops("rejected-aggregated") else {i for b in buckets.values() for i, _ in b["ids"]}
|
|
stated = {who: sum(len(v - aggregated) for t, v in by.items() if q[0] <= t < q[0] + q[1]) # §6.1: none of them aggregated
|
|
for who, by in rejected_by.items()}
|
|
check_collection(doc, s, cfg, tid, req, q, mg, csk, task, bad, stated)
|
|
for rid, x in listed.items():
|
|
rep = first.get(rid)
|
|
if rep is None:
|
|
bad("report %s is never uploaded" % rid)
|
|
continue
|
|
if rep["m"]["time"] != x["time"]:
|
|
bad("report %s: its time" % rid)
|
|
if rep["leader"]["config_id"] != lconf["id"] or rep["helper"]["config_id"] != hconf["id"]:
|
|
bad("report %s is not sealed to each list's first configuration of DAP's suite (§6.3)" % rid)
|
|
if rid in late:
|
|
want = ("report_replayed", None)
|
|
elif x["kind"] == "presentation":
|
|
want = "committed"
|
|
else:
|
|
want = (x["error"], x["reason"])
|
|
if fate.get(rid) != want:
|
|
bad("report %s (%s): stated %s, the run gives %s" % (rid, x["kind"], want, fate.get(rid)))
|
|
if x["kind"] == "presentation":
|
|
check_presentation(doc, f, x, rep, tf, cat, task, lsk, hsk, ctx, bad)
|
|
for rid, rep in first.items():
|
|
if rid not in listed:
|
|
bad("an uploaded report %s is not listed" % rid)
|
|
if outcomes_stated is None:
|
|
bad("no aggregation job")
|
|
return fails
|
|
|
|
|
|
def check_presentation(doc, f, x, rep, tf, cat, task, lsk, hsk, ctx, bad):
|
|
"""A presentation's report: §6.2's contribution, §6.3's extensions and its one-hot measurement."""
|
|
p, tid, m = task["p"], task["tid"], rep["m"]
|
|
aad = TAV.input_share_aad(tid, task["cfg"], m, rep["public"])
|
|
try:
|
|
lexts, lpay = open_input_share(lsk, 2, aad, rep["leader"])
|
|
hexts, hpay = open_input_share(hsk, 3, aad, rep["helper"])
|
|
except ValueError:
|
|
bad("presentation %s does not open" % x["id"])
|
|
return
|
|
if m["exts"] or lexts != hexts or (cat["attest"] and [e[0] for e in lexts] != [TOKEN_EXT]) or \
|
|
(not cat["attest"] and lexts):
|
|
bad("presentation %s: its extensions are not §6.3's" % x["id"])
|
|
return
|
|
pool = [{"id": x["id"], "token": lexts[0][1].hex()}] if cat["attest"] else []
|
|
record = bytes.fromhex(doc["fixtures"]["records"][x["record"]]["bytes"])
|
|
region = None if x["region"] is None else x["region"].encode("utf-8")
|
|
got = contribute(doc, cat, task, record, [bytes.fromhex(x["claim"])], region, x["at_ms"], pool)
|
|
want = {"outcome": "report", "cell": x["cell"], "creative": x["creative"], "column": x["column"],
|
|
"quartile": x["quartile"], "time": x["time"], "id": x["id"] if cat["attest"] else "fresh", "discarded": []}
|
|
if got != want:
|
|
bad("presentation %s: its contribution (§6.2) is %s" % (x["id"], json.dumps(got)))
|
|
meas = TAV.vadd(p.f, meas_share(p, ctx, 0, lpay), meas_share(p, ctx, 1, hpay))
|
|
if meas != [1 if k == x["cell"] else 0 for k in range(cat["cells"])]:
|
|
bad("presentation %s: its measurement is not one at its cell (§6.3)" % x["id"])
|
|
|
|
|
|
def check_collection(doc, s, cfg, tid, req, q, mg, csk, task, bad, stated):
|
|
"""A completed collection: the shares, the statements, the result and the receipt."""
|
|
p, info = task["p"], task["info"]
|
|
share_req = req + b"\x01" + (16).to_bytes(2, "big") + q[0].to_bytes(8, "big") + q[1].to_bytes(8, "big") + \
|
|
mg["n"].to_bytes(8, "big") + mg["ck"]
|
|
if share_req.hex() != s["share_req"]:
|
|
bad("collection %s: the AggregateShareReq is not DAP §4.6.4's" % s["name"])
|
|
aad = TAV.agg_share_aad(tid, cfg, req)
|
|
try:
|
|
hct = TAV.read_ct(TAV.Rd(bytes.fromhex(s["helper_share"])))
|
|
hshare = TAV.hpke_open(hct["enc"], csk, b"dap-20 aggregate share" + bytes([3, 0]), aad, hct["payload"])
|
|
r = TAV.Rd(bytes.fromhex(s["resp"]))
|
|
n, iv = r.u(8), (r.u(8), r.u(8))
|
|
lct, hct2 = TAV.read_ct(r), TAV.read_ct(r)
|
|
r.end()
|
|
lshare = TAV.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 = TAV.read_ct(TAV.Rd(bytes.fromhex(s[k])))
|
|
info_s = b"wai-telemetry-aggregate statement" + bytes([role, 0])
|
|
sts.append(TAV.read_statement(TAV.hpke_open(sct["enc"], csk, info_s, aad, sct["payload"])))
|
|
except (ValueError, Refuse):
|
|
bad("collection %s does not open with the collector's key" % s["name"])
|
|
return
|
|
times = mg["times"]
|
|
if n != mg["n"] or iv != (min(times), max(times) - min(times) + 1) or hct2 != hct:
|
|
bad("collection %s: the CollectionJobResp's count, interval or helper share" % s["name"])
|
|
if TAV.decode_vec(p.f, lshare) != mg["l"] or TAV.decode_vec(p.f, hshare) != mg["h"]:
|
|
bad("collection %s: an aggregate share is not the batch's" % s["name"])
|
|
for st, role, share, key, who in ((sts[0], 2, lshare, info["leader"], "leader"), (sts[1], 3, hshare, info["helper"], "helper")):
|
|
if (st["role"], st["task_id"], st["collection"], st["batch"], st["count"], st["checksum"], st["agg_share"],
|
|
st["signer"]) != (role, tid, sha(req), q, mg["n"], mg["ck"], share, key):
|
|
bad("collection %s: a statement is not its share's (telemetry-aggregation §6.2)" % s["name"])
|
|
if st["rejected"] != stated[who]: # §7.3: what it rejected, of this batch
|
|
bad("collection %s: the %s states %s rejected, the run gives %s" % (s["name"], who, st["rejected"], stated[who]))
|
|
if ED.verify(key, b"wai:telemetry-aggregate/share\x01" + st["body"], st["sig"]) is not None:
|
|
bad("collection %s: a statement's signature" % s["name"])
|
|
result = TAV.vadd(p.f, mg["l"], mg["h"])
|
|
if result != s["result"]:
|
|
bad("collection %s: the result" % s["name"])
|
|
rc = TAV.read_receipt(bytes.fromhex(s["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, req, iv):
|
|
bad("collection %s: the receipt is not the collection's" % s["name"])
|
|
given = sorted(json.dumps({"id": rid.hex(), "time": tm, "task": tid.hex()}, sort_keys=True)
|
|
for rid, tm in mg["ids"])
|
|
holders = [x for x in doc["receipts"] if x["receipt"] == s["receipt"]]
|
|
with_reports = [x for x in holders if x["reports"] is not None]
|
|
if not holders or (with_reports and not any(sorted(json.dumps(g, sort_keys=True) for g in x["reports"]) == given
|
|
for x in with_reports)):
|
|
bad("collection %s: the receipts section does not hold its receipt with its reports" % s["name"])
|
|
|
|
|
|
# ------------------------------------------------------------------ the corpus
|
|
|
|
KINDS = ("catalogues", "tasks", "task_files", "contributions", "uploads", "plans", "settles", "receipts", "tables")
|
|
|
|
|
|
def run(doc, kind, c):
|
|
if kind == "catalogues":
|
|
return catalogue_outcome(bytes.fromhex(c["bytes"]))
|
|
if kind == "tasks":
|
|
return task_outcome(bytes.fromhex(c["task"]), bytes.fromhex(c["catalogue"]))
|
|
if kind == "task_files":
|
|
return task_file_outcome(doc, c)
|
|
if kind == "contributions":
|
|
return contribution_outcome(doc, c)
|
|
if kind == "uploads":
|
|
return upload_outcome(bytes.fromhex(c["bytes"]))
|
|
if kind == "plans":
|
|
return plan_outcome(doc, c)
|
|
if kind == "settles":
|
|
return settle(bytes.fromhex(c["sent"]), bytes.fromhex(c["response"]))
|
|
if kind == "receipts":
|
|
return receipt_outcome(c)
|
|
return table(bytes.fromhex(c["catalogue"]), bytes.fromhex(c["task"]), c["receipts"])
|
|
|
|
|
|
def check_fixtures(doc):
|
|
"""What the fixtures say of themselves: each creative's digest (ad-presentation §4), each catalogue,
|
|
task and task file the cases name by fixture."""
|
|
fails = []
|
|
fx = doc["fixtures"]
|
|
for n, c in fx["creatives"].items():
|
|
objs = [{"content_hash": bytes.fromhex(o["content_hash"]), "duration_ms": o["duration_ms"]}
|
|
for o in c["objects"]]
|
|
if APV.creative_digest(objs).hex() != c["digest"]:
|
|
fails.append("creative %s: its digest" % n)
|
|
for n, h in fx["task_files"].items():
|
|
try:
|
|
tf, cat, task, _ = receiver_task_file(doc, n)
|
|
if tf["catalogue"].hex() != fx["catalogues"].get(n, tf["catalogue"].hex()) or \
|
|
tf["task"].hex() != fx["tasks"].get(n, tf["task"].hex()):
|
|
fails.append("task file %s: not its fixtures' catalogue and task" % n)
|
|
except Refuse as e:
|
|
fails.append("task file %s is refused %s under what the receiver holds" % (n, e.reason))
|
|
return fails
|
|
|
|
|
|
def strings(v, out):
|
|
if isinstance(v, str):
|
|
out.append(v)
|
|
elif isinstance(v, list):
|
|
for x in v:
|
|
strings(x, out)
|
|
elif isinstance(v, dict):
|
|
for x in v.values():
|
|
strings(x, out)
|
|
|
|
|
|
def stated_reasons(doc):
|
|
"""Every reason an outcome of the corpus gives, flows included."""
|
|
out = []
|
|
for kind in KINDS:
|
|
for c in doc[kind]:
|
|
o = c["outcome"]
|
|
if isinstance(o, dict) and o.get("reason"):
|
|
out.append(o["reason"])
|
|
for s in [] if kind != "tables" or o.get("outcome") != "table" else o["receipts"]:
|
|
m = re.search(r"reason=(\S+)", s["verdict"])
|
|
if m:
|
|
out.append(m.group(1))
|
|
for f in doc["flows"]:
|
|
for x in f["reports"]:
|
|
out += [y for y in (x.get("error"), x.get("reason")) if y]
|
|
for s in f["steps"]:
|
|
out += [e["error"] for e in s.get("errors", [])]
|
|
out += [y for o in s.get("outcomes", []) for y in (o["outcome"], o["reason"]) if y and y != "committed"]
|
|
if s["step"] == "collect":
|
|
out += [y for y in (s["outcome"], s.get("helper_outcome"))
|
|
if y not in ("processing", "completed", "accept", None)]
|
|
return out
|
|
|
|
|
|
ELSEWHERE = set(TAV.REASONS) | set(SPV.REASONS) | set(SPV.OTHER) | set(APV.REASONS) | set(JWP_REASONS) | \
|
|
set(REPORT_ERRORS.values()) | set(PROBLEMS) | {"count_mismatch", "truncated", "malformed", "unknown_version",
|
|
"bad_key", "bad_signature"}
|
|
REUSED = ["count_mismatch", "attestation_absent", "attestation_invalid", "task_invalid", "measure_mismatch",
|
|
"floor_too_low", "aggregator_alone", "batch_invalid", "task_mismatch", "batch_too_small", "batch_mismatch",
|
|
"bad_key", "bad_signature", "pod_signer_unknown", "claim_not_viewable", "claim_duplicate",
|
|
"claim_marker_absent", "claim_manifest_unknown", "signer_id_not_origin_bound", "origin_not_anchored",
|
|
"authority_not_granted", "anchor_not_pinned", "key_revoked", "report_replayed", "report_dropped",
|
|
"invalid_message", "hpke_decrypt_error", "vdaf_verify_error", "invalidBatchSize"]
|
|
|
|
|
|
def check(doc, quiet=False):
|
|
"""The failures of a corpus document."""
|
|
fails = check_fixtures(doc)
|
|
for kind in KINDS:
|
|
for c in doc[kind]:
|
|
try:
|
|
got = run(doc, kind, c)
|
|
except (Refuse, KeyError, TypeError, ValueError, IndexError, AttributeError) as e:
|
|
got = {"error": repr(e)}
|
|
if got != c["outcome"]:
|
|
fails.append("%s %s" % (kind, c["name"]))
|
|
if not quiet:
|
|
print("FAIL %s %s\n stated %s\n got %s" % (kind, c["name"], json.dumps(c["outcome"])[:600],
|
|
json.dumps(got)[:600]))
|
|
for f in doc["flows"]:
|
|
try:
|
|
ff = run_flow(doc, f)
|
|
except (Refuse, KeyError, TypeError, ValueError, IndexError, AttributeError) as e:
|
|
ff = ["flow %s: %r" % (f["name"], e)]
|
|
fails += ff
|
|
if not quiet:
|
|
for x in ff:
|
|
print("FAIL", x)
|
|
if not ff:
|
|
n = len(f["reports"]) + len(f.get("late", []))
|
|
print("PASS flow %s: %d reports re-run through %d steps" % (f["name"], n, len(f["steps"])))
|
|
if doc.get("codes") != REASONS:
|
|
fails.append("the corpus's codes are not §10's")
|
|
given = stated_reasons(doc)
|
|
for code in REASONS + REUSED:
|
|
if code not in given:
|
|
fails.append("no case gives " + code)
|
|
for code in sorted(set(given) - set(REASONS) - ELSEWHERE):
|
|
fails.append("a case gives %s, a code the extension does not name" % code)
|
|
return fails
|
|
|
|
|
|
def bounds():
|
|
"""§5.4's and §6.4's size bounds, which no vector carries: a file of exactly the bound that
|
|
reads in every other respect is read (a task file passes step 1 and is refused at step 2, its
|
|
catalogue of zeros not reading), and the same file one byte longer is refused at step 1.
|
|
§4.3's bound decides no outcome: no catalogue that meets §4.3 comes near it."""
|
|
fails = []
|
|
suite = struct.pack(">BHHHH", 1, *SUITE, 32) + b"\x09" * 32
|
|
|
|
def hpke_list(cid):
|
|
body = bytes([cid]) + suite[1:]
|
|
return len(body).to_bytes(2, "big") + body
|
|
|
|
def task_file(n):
|
|
body = (TASK_MAGIC + b"\x01" + (0).to_bytes(2, "big") + u32(n) + b"\x00" * n + hpke_list(1) + hpke_list(2)
|
|
+ b"\x00" * 32 + (1).to_bytes(2, "big") + b"x")
|
|
return body + b"\x00" * 64
|
|
|
|
ct = b"\x01" + (1).to_bytes(2, "big") + b"\x01" + u32(1) + b"\x01"
|
|
|
|
def upload(n):
|
|
report = b"\x00" * 16 + (0).to_bytes(8, "big") + (0).to_bytes(2, "big") + u32(n) + b"\x00" * n + ct + ct
|
|
return write_upload(b"\x00" * 32, [report])
|
|
|
|
def outcome(fn, b):
|
|
try:
|
|
fn(b)
|
|
return "read"
|
|
except Refuse as e:
|
|
return e.reason
|
|
|
|
for what, make, bound, fn, inside, past in (
|
|
("task file", task_file, MAX_TASK_FILE, lambda b: accept_task_file(b, [], []), "catalogue_malformed",
|
|
"task_file_malformed"),
|
|
("upload file", upload, MAX_UPLOAD, read_upload, "read", "upload_malformed")):
|
|
at = bound - len(make(0))
|
|
for b, want in ((make(at), inside), (make(at + 1), past)):
|
|
got = outcome(fn, b)
|
|
if got != want:
|
|
fails.append("a %s of %d bytes: %s, not %s" % (what, len(b), got, want))
|
|
return fails
|
|
|
|
|
|
# ------------------------------------------------------------------ mutations and weakenings
|
|
|
|
def flip(h, at):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at % len(b)] ^= 0x01
|
|
return b.hex()
|
|
|
|
|
|
def case(d, kind, name):
|
|
return next(c for c in d[kind] if c["name"] == name)
|
|
|
|
|
|
def flow(d, name):
|
|
return next(f for f in d["flows"] if f["name"] == name)
|
|
|
|
|
|
def step(f, name):
|
|
return next(s for s in f["steps"] if s.get("name") == name or s["step"] == name)
|
|
|
|
|
|
def mutations():
|
|
"""Copies of the corpus, one defect each, every one of which must fail."""
|
|
def m_catalogue_hold(d):
|
|
c = case(d, "catalogues", "flight")
|
|
b = bytearray.fromhex(c["bytes"])
|
|
b[5 + 4 + 2 + 14 + 3] ^= 0x01 # the last byte of hold_s
|
|
c["bytes"] = b.hex()
|
|
|
|
def m_region_column(d):
|
|
case(d, "contributions", "region_subdivision_of_listed_country")["outcome"].update(column="other", cell=14)
|
|
|
|
def m_bucket_edge(d):
|
|
case(d, "contributions", "bucket_first_ms")["at_ms"] -= 1
|
|
|
|
def m_quartile(d):
|
|
case(d, "contributions", "q4_three_quarters")["outcome"].update(quartile=5, cell=4)
|
|
|
|
def m_pool_discarded(d):
|
|
case(d, "contributions", "token_pool_skips_bad")["outcome"]["discarded"] = []
|
|
|
|
def m_task_retention(d):
|
|
case(d, "tasks", "flight")["outcome"]["retention_s"] += 1
|
|
|
|
def m_task_file_bit(d):
|
|
c = case(d, "task_files", "flight")
|
|
c["bytes"] = flip(c["bytes"], 40)
|
|
|
|
def m_anchor(d):
|
|
case(d, "task_files", "flight")["anchors"] = ["fetched"]
|
|
|
|
def m_floor_flag(d):
|
|
case(d, "tables", "flight")["outcome"]["totals"]["rows"][0]["floor_met"] = False
|
|
|
|
def m_row_count(d):
|
|
case(d, "tables", "flight")["outcome"]["receipts"][1]["rows"][5]["presentations"] = 50
|
|
|
|
def m_receipt_byte(d):
|
|
c = case(d, "receipts", "r0")
|
|
c["receipt"] = flip(c["receipt"], len(c["receipt"]) // 4)
|
|
|
|
def m_receipt_reports(d):
|
|
case(d, "receipts", "r0")["reports"].pop()
|
|
|
|
def m_hold(d):
|
|
s = step(flow(d, "flight"), "collect_b0_held")
|
|
s["now_s"] += 1
|
|
|
|
def m_late_error(d):
|
|
step(flow(d, "flight"), "late_b0")["errors"][0]["error"] = "batch_collected"
|
|
|
|
def m_token_outcome(d):
|
|
o = step(flow(d, "flight"), "aggregate")["outcomes"][0]
|
|
o.update(outcome="committed", reason=None)
|
|
|
|
def m_init_byte(d):
|
|
s = step(flow(d, "flight"), "aggregate")
|
|
s["init_req"] = flip(s["init_req"], 400)
|
|
|
|
def m_merge_share(d):
|
|
s = step(flow(d, "flight"), "relay_merged")
|
|
s["shares"].reverse()
|
|
|
|
def m_presentation_cell(d):
|
|
x = next(r for r in flow(d, "open")["reports"] if r["kind"] == "presentation")
|
|
x["cell"] = (x["cell"] + 1) % 5
|
|
|
|
def m_budget(d):
|
|
case(d, "plans", "budget_exact")["budget"] -= 1
|
|
|
|
def m_plan_retention(d):
|
|
case(d, "plans", "retention_exact")["t_ms"] -= 1
|
|
|
|
def m_upload_order(d):
|
|
c = case(d, "uploads", "three_reports")
|
|
tid, reps = read_upload(bytes.fromhex(c["bytes"]))
|
|
c["bytes"] = write_upload(tid, [reps[1]["raw"], reps[0]["raw"], reps[2]["raw"]]).hex()
|
|
|
|
def m_settle_too_early(d):
|
|
o = case(d, "settles", "each_error")["outcome"]
|
|
o["delete"] = sorted(o["delete"] + o["keep"])
|
|
o["keep"] = []
|
|
|
|
def m_rejected_outcome(d):
|
|
case(d, "receipts", "r0")["outcome"]["rejected"]["leader"] += 1
|
|
|
|
def m_rejected_no_count(d):
|
|
case(d, "receipts", "r0_of_version_1")["outcome"]["rejected"]["helper"] = 0
|
|
|
|
def m_rejected_row(d):
|
|
e = case(d, "tables", "flight")["outcome"]["receipts"][0]["rejected"]
|
|
e["leader"], e["helper"] = e["helper"], e["leader"]
|
|
|
|
def m_rejected_total(d):
|
|
case(d, "tables", "flight")["outcome"]["totals"]["rejected"]["helper"] += 1
|
|
|
|
def m_rejected_total_of_none(d):
|
|
case(d, "tables", "flight_a_statement_of_no_count")["outcome"]["totals"]["rejected"]["leader"] = 0
|
|
|
|
def m_rejected_of_both_versions(d):
|
|
case(d, "tables", "flight_one_statement_of_no_count")["outcome"]["totals"]["rejected"]["helper"] = 3
|
|
|
|
def m_replay_answer(d):
|
|
s = next(x for x in case(d, "flows", "flight")["steps"] if x["step"] == "replay")
|
|
s["resp"] = s["resp"][:-2] + "00"
|
|
|
|
def m_overlap_gone(d):
|
|
d["tables"].remove(case(d, "tables", "flight_overlap"))
|
|
|
|
def m_codes(d):
|
|
d["codes"].reverse()
|
|
|
|
def m_creative_fixture(d):
|
|
o = d["fixtures"]["creatives"]["A"]["objects"][0]
|
|
o["duration_ms"] += 1
|
|
|
|
return [m_catalogue_hold, m_region_column, m_bucket_edge, m_quartile, m_pool_discarded, m_task_retention,
|
|
m_task_file_bit, m_anchor, m_floor_flag, m_row_count, m_receipt_byte, m_receipt_reports, m_hold,
|
|
m_late_error, m_token_outcome, m_init_byte, m_merge_share, m_presentation_cell, m_budget,
|
|
m_plan_retention, m_upload_order, m_settle_too_early, m_overlap_gone, m_codes, m_creative_fixture,
|
|
m_rejected_outcome, m_rejected_no_count, m_rejected_row, m_rejected_total, m_rejected_total_of_none,
|
|
m_rejected_of_both_versions, m_replay_answer]
|
|
|
|
|
|
WEAKENINGS = {
|
|
"cat-version": "§4.3 rule 1: version 1",
|
|
"cat-terms": "§4.3 rule 2: the terms",
|
|
"cat-hold": "§4.3 rule 3: a hold above zero",
|
|
"cat-signers": "§4.3 rule 4: one to sixteen signers",
|
|
"cat-creatives": "§4.3 rule 5: a creative",
|
|
"cat-order": "§4.3 rules 4 to 8: strictly ascending lists",
|
|
"cat-region": "§4.3 rule 6: each region a code",
|
|
"cat-overlap": "§4.3 rule 6: no country beside its subdivision",
|
|
"cat-cells": "§4.3 rule 7: at most 4,096 cells",
|
|
"cat-keys": "§4.3 rule 8: the keys' count",
|
|
"cat-key-form": "§4.3 rule 8: each key's form",
|
|
"catalogue-digest": "§5.3 step 2: measure 5 and the catalogue's digest",
|
|
"task-precision": "§5.3 step 3: an hour's precision",
|
|
"task-interval": "§5.3 step 3: a flight of a duration",
|
|
"task-retention": "§5.3 step 3: the retention instant's bound",
|
|
"chunk-length": "§5.3 step 4: the chunk length",
|
|
"tf-version": "§5.4 step 1: version 1",
|
|
"tf-signer-ascii": "§5.4 step 1: an ASCII signer_id",
|
|
"tf-hpke-ids": "§5.4 step 1: distinct configuration ids",
|
|
"tf-hpke-suite": "§5.4 step 1: DAP's suite in each list",
|
|
"tf-signature": "§5.4 step 4: the file's signature",
|
|
"tf-authority": "§5.4 step 5: the authority aggregate/task",
|
|
"tf-pinned": "§5.4 step 5: a pinned anchor",
|
|
"tf-revocation": "§5.4 step 5: revisions the bearer brings",
|
|
"shared": "§6.2 step 2: a shared record",
|
|
"terms-equal": "§6.2 step 3: the catalogue's terms",
|
|
"creative-listed": "§6.2 step 5: a listed creative",
|
|
"instant": "§6.2 step 6: a known instant",
|
|
"flight-window": "§6.2 step 7: a time within the flight",
|
|
"token-pool": "§6.2 step 8: each pair's token",
|
|
"region-subdivision": "§4.2: a subdivision's column is its listed country's",
|
|
"hpke-first": "§6.3: the first configuration of DAP's suite",
|
|
"token-check": "§7.2: the aggregators' report rule",
|
|
"token-challenge": "§7.2: a token's challenge",
|
|
"helper-token": "§7.2: the helper checks its own share",
|
|
"public-extension": "§7.2: no public extension",
|
|
"hold": "§7.1: the hold",
|
|
"upload-dropped": "DAP §4.4.2.2: a report outside the flight dropped at upload",
|
|
"upload-replay": "DAP §4.4.2.2: a report replayed at upload",
|
|
"batch-in-task": "§8.1 step 12: the batch within the flight",
|
|
"count-sum": "§8.1 step 13: the elements sum to the count",
|
|
"count-integer": "§8.1 step 13: integers, each at most the count",
|
|
"task-equal": "§8.2 step 2: the given task",
|
|
"dedupe": "§8.2 step 3: equal receipts are one",
|
|
"overlap": "§8.2 step 3: overlapping receipts set aside",
|
|
"reports-checked": "§8.2: whether a receipt's reports were checked",
|
|
"rejected-own": "§7.3: each aggregator states its own count",
|
|
"rejected-aggregated": "telemetry-aggregation §6.1: no report it aggregated counts as rejected",
|
|
"rejected-none": "§8.2: a total of none where a statement states no count",
|
|
"checked-any": "§8.2: a receipt given more than once checked when any occurrence was",
|
|
"tf-bound": "§5.4 step 1: at most 1,048,576 bytes",
|
|
"upload-bound": "§6.4: at most 16,777,216 bytes",
|
|
"upload-version": "§6.4: version 1",
|
|
"upload-length": "§6.4: a file that ends before its upload's length",
|
|
"upload-order": "§6.4: reports in ascending order of id",
|
|
"plan-deadline": "§6.5: timely and late",
|
|
"plan-retention": "§6.5: dropped at the retention instant",
|
|
"plan-budget": "§6.5: the budget",
|
|
"settle-subsequence": "§6.6: a response of the file's ids in order",
|
|
"settle-too-early": "§6.6: report_too_early kept",
|
|
"settle-unknown-value": "§6.6: a value DAP does not define kept",
|
|
}
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) != 2:
|
|
print(__doc__)
|
|
return 2
|
|
fails = []
|
|
st = ED._self_test()
|
|
ct = TAV.crypto_self_test()
|
|
print(("PASS" if not st and not ct else "FAIL") + " Ed25519, TurboSHAKE128, X25519 and HPKE self-tests")
|
|
fails += (["ed25519 self-test"] if st else []) + ct
|
|
print(("PASS" if len(REASONS) == 11 else "FAIL") + " §10 read from the extension: %d codes" % len(REASONS))
|
|
with open(os.path.join(argv[1], "vectors.json"), encoding="utf-8") as fh:
|
|
doc = json.load(fh)
|
|
got = check(doc)
|
|
fails += got
|
|
sizes = ", ".join("%d %s" % (len(doc[k]), k) for k in KINDS + ("flows",))
|
|
print("%s %s: %d failure(s)" % ("PASS" if not got else "FAIL", sizes, len(got)))
|
|
extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
|
|
if extra:
|
|
fails.append("files the corpus does not hold: %s" % extra)
|
|
b = bounds()
|
|
print(("PASS" if not b else "FAIL") + " the size bounds of §5.4 and §6.4, at their edge")
|
|
fails += b
|
|
caught = 0
|
|
for m in mutations():
|
|
d = copy.deepcopy(doc)
|
|
m(d)
|
|
if check(d, quiet=True):
|
|
caught += 1
|
|
else:
|
|
fails.append("mutation %s was not caught" % m.__name__)
|
|
n = len(mutations())
|
|
print("%s %d of %d mutated copies fail" % ("PASS" if caught == n else "FAIL", caught, n))
|
|
weak = 0
|
|
for name, what in WEAKENINGS.items():
|
|
SKIP.clear()
|
|
SKIP.add(name)
|
|
if check(doc, quiet=True) or bounds():
|
|
weak += 1
|
|
else:
|
|
fails.append("a verifier without %s (%s) passes the corpus" % (what, name))
|
|
SKIP.clear()
|
|
n = len(WEAKENINGS)
|
|
print("%s %d of %d weakened verifiers are caught" % ("PASS" if weak == n else "FAIL", weak, n))
|
|
for x in fails:
|
|
print("FAIL", x)
|
|
print(("PASS" if not fails else "FAIL") + " impression aggregates corpus")
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|