Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 17 Packages Wiki Activity Actions 180
open-standards/wai/live-conformance/age_assurance_verify.py
Transaction Science 298eaa70f5 age × co: the room's own service is not held to the serving rule 
A room with both `[age]` and `[co]` broke: co-performance's clock follows and
logs every performer's objects over a relay session of its own, which holds no
grant and no token, so age-assurance §5.5 served it nothing rated.

- age-assurance §5.5: a session that proves possession of a room key (the
  authority's or the clock's) is the room's own service, served every object of
  the room. The proof is a GRANT_PROOF naming no grant (32 bytes of zero) over
  the session's challenge, checked as live-rooms §10.5.4 checks a proof; one
  that does not pass refuses the session with that section's reason; one that
  passes by a key that is no room key gives nothing. Co-performance §8 points
  to it; §11 and §12 name the reasons and the new vectors.
- joule-moq: `grant_proof::NO_GRANT`, `add_room_key_proof`, `room_key_proofs`.
- joule-moq-relay: SETUP checks room-key proofs; a session proving a key the
  relay holds as a room key (an authority's, for its rooms; `[floor]` or `[dvr]`
  room keys, for every room) has `SessionCtx::service`, and its threshold there
  is past every age. `ClientOptions::room_keys` presents such proofs.
  tests/age_room_service.rs: a room rated 18 with the floor held; the clock's
  and the authority's proofs served every group, the clock's token alone and a
  non-room key refused `below_threshold`, a forged clock proof and one over
  another challenge closed UNAUTHORIZED.
- joule-moq-room: the clock's session proves both room keys it holds.
- wai-rs `age_assurance::room_service`, the corpus case `room_service`, and
  `age_assurance_verify.py` taught the rule, with three new checks and four
  mutations, all caught.
- e2e: scenario `ageco`, a room with both `[age]` and `[co]`.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-22 07:08:25 -04:00

1602 lines
72 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""Second implementation of extensions/age-assurance.md (`wai.age.assurance`), over the corpus
`live-conformance/age/`.
Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. Everything the extension defines is written here:
- reading a record (§2): a JSON reader of its own (RFC 8259's grammar, no byte-order mark, no
unpaired surrogate, no member named twice, at most 32 levels, as jwp-receipts §3.1 step 1
reads a claim), then the encodings of jwp-receipts §2.4.3 (no boolean, no number but an
unsigned integer of at most 2^53 - 1, terms, absolute URIs, date-times), exact member sets,
JCS (RFC 8785, as jwp-receipts §2.4.3 reduces it), the signature domain and the link hash;
- rating claims (§3.1, §3.2), coverage and ages (§3.3), carrying through a derivation (§3.4,
input by input and rule by rule), and whose claims a relay honours, refuses or, unable to
attribute them to the room, passes over (§3.6);
- the age policy (§4), with its issuer keys read as RFC 9578 §6.5's DER SubjectPublicKeyInfo:
the AlgorithmIdentifier of RFC 9578 Appendix A.2's key, byte for byte, and an RSAPublicKey
of a 2048-bit modulus and a positive exponent;
- the TokenChallenge (RFC 9577 §2.1.1) and the token (RFC 9577 §2.2.1, RFC 9578 §6), the six
steps of §5.2 in their order, and RSASSA-PSS-VERIFY (RFC 8017 §8.1.2, EMSA-PSS-VERIFY of
§9.1.2 and MGF1 of Appendix B.2.1) with SHA-384 and a 48-byte salt, as RFC 9578 §6.4 has it;
- serving a rated track (§5.5), with the room's own service a session's proof of a room key
makes (live-rooms §10.5.4, the grant's digest 32 bytes of zero), and the sink's age state and
predicate (§6);
- the admission receipt (§7): §7.3's reading, `receipt_identifies` first, its checks in their
order, and the cells §7.2 writes from the admissions;
- the grant attribute `age` (§5.4) and the minors' profile (§8.1, §8.2): the descriptor's rules 1
to 3 at the first record that puts a minor's publishing grant into effect, then rule 4 over
every grant.
Every Ed25519 signature is verified with WAI's strict rule (claim-conformance/_ed25519.py),
every link hash with BLAKE3 written from its specification (claim-conformance/_blake3.py), and
each room by the live-rooms corpus's second implementation (live_rooms_verify.py). §11's codes
are read from the extension.
python3 -I age_assurance_verify.py <live-conformance/age>
It gives every case the outcome, the reason, the place and every statement the corpus states;
checks RFC 9578 Appendix A.2's first token vector; checks that every reason of §11 refuses a
case; then mutates copies of the corpus and requires every copy to fail, and runs the corpus
under copies of this verifier with one of its own checks left out, and requires the corpus to
catch each one it covers. Exit status 0 only if all of that holds. Python 3.8 or later;
standard library only.
"""
import copy
import hashlib
import json
import os
import re
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import live_events_verify as LEV # noqa: E402
import live_rooms_verify as LRV # noqa: E402
import _ed25519 as ED # noqa: E402
from _blake3 import blake3 # noqa: E402
EXTENSION = os.path.join(HERE, "..", "extensions", "age-assurance.md")
SIGN_DOMAIN = b"wai:age-assurance\x01" # §2
LINK_DOMAIN = b"wai:age-assurance-link\x01" # §2
MAX_INT = (1 << 53) - 1 # jwp-receipts §2.4.3
MAX_DEPTH = 32 # jwp-receipts §3.1 step 1
MAX_AGE = 21 # §2: an age is 0 to 21
FLOOR = 50 # §7.2: a cell under 50 is suppressed
TOKEN_TYPE = b"\x00\x02" # RFC 9578 §6
TOKEN_LEN = 2 + 32 + 32 + 32 + 256 # §5.2 step 1: 354
SIGNED_LEN = 2 + 32 + 32 + 32 # §5.2 step 4: the token's first 98 bytes
SALT_LEN = 48 # RFC 9578 §6.4
CHAT, REACTION = 0x03, 0x04 # live-events §4.4, §4.5
RECORD_POWER, SEEK_POWER = 1 << 3, 1 << 4 # live-rooms §3.3
JOIN, EXCHANGE = 0x10, 0x14 # live-events §4, live-rooms §7.2
# §4.2: the AlgorithmIdentifier of RFC 9578 Appendix A.2's key, byte for byte: id-RSASSA-PSS
# (1.2.840.113549.1.1.10) with RSASSA-PSS-params of hashAlgorithm SHA-384
# (2.16.840.1.101.3.4.2.2), maskGenAlgorithm MGF1 (1.2.840.113549.1.1.8) with SHA-384, and
# saltLength 48.
PSS_ALGORITHM = bytes.fromhex(
"303d"
"06092a864886f70d01010a"
"3030"
"a00d300b0609608648016503040202"
"a11a301806092a864886f70d010108300b0609608648016503040202"
"a203020130")
CLAIM = ["kind", "subject", "ratings", "inputs", "record", "issued_at", "signer_pubkey", "sig"] # §3.1
POLICY = ["kind", "room", "issuers", "window_s", "issued_at", "signer_pubkey", "sig"] # §4.1
RECEIPT = ["kind", "room", "policy", "start", "end", "window_s", "cells", "issued_at", "signer_pubkey", "sig"] # §7.1
CELL = ["window", "issuer", "threshold", "admitted", "suppressed"] # §7.1
HEX64 = re.compile(r"[0-9a-f]{64}")
HEX = re.compile(r"(?:[0-9a-f]{2})+")
AGE_ATTR = re.compile(r"[1-9][0-9]?")
URI_SCHEME = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*:.+", re.S)
DATE_TIME = re.compile(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(\.[0-9]+)?"
r"([Zz]|[+-]([0-9]{2}):([0-9]{2}))")
# Unicode's White_Space property (a term holds none of these, jwp-receipts §2.4.3).
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0          "
"  

  　")
# ---------------------------------------------------------------- the checks a self-test drops
# Every check of this verifier that a weakened copy may leave out, with whether the corpus is
# meant to catch its absence (§12 lists a case for it). A check the corpus does not reach is
# reported, not failed.
CHECKS = {
# §2, jwp-receipts §2.4.3 and §3.1 step 1
"json_duplicate": (True, "no member named twice"),
"json_depth": (False, "at most 32 levels"),
"json_encodings": (True, "no boolean read as a number, no number but an unsigned integer"),
"exact_members": (True, "each record has exactly its members"),
"kind": (False, "a record's kind names it"),
"date_time": (True, "issued_at is a date-time"),
"signature": (True, "a record's signature verifies"),
"strict_key": (True, "a signer key of small order is bad_key"),
# §3.1, §3.2
"subject_size": (True, "a digest or an id of a subject is 32 bytes"),
"span_order": (True, "a span's to_group is not below its from_group"),
"ratings_count": (False, "1 to 8 ratings"),
"ratings_order": (True, "ratings in strictly ascending order of scheme"),
"absolute_uri": (True, "a scheme is an absolute URI"),
"value_bytes": (True, "a value is 1 to 64 bytes of ! to ~"),
"min_age": (True, "a min_age is at most 21"),
"inputs_count": (False, "at most 16 inputs"),
"inputs_distinct": (False, "no input's content hash twice"),
"consistency": (True, "one scheme's one value means one age"),
# §3.4
"rule1_held": (True, "rule 1: an input's rating names a claim the verifier holds"),
"rule1_subject": (True, "rule 1: an object claim an input names is of that input"),
"rule2": (True, "rule 2: a rated input names a claim"),
"rule3_scheme": (True, "rule 3: every scheme of an input's claims is carried"),
"rule3_age": (True, "rule 3: at an age at least as great"),
# §3.3, §3.6, §5.5
"track_from": (True, "a track claim covers from its from_group on"),
"span_end": (True, "a span claim covers up to its to_group"),
"honour_room_signer": (True, "a room claim is honoured only from a room key"),
"honour_track_signer": (True, "a track's claim is honoured only from a room key or its owner"),
"room_claim_room": (True, "a room claim names the room"),
"attribution": (True, "a claim the relay cannot attribute is neither honoured nor refused"),
"serve_rule": (True, "a request is served only at a threshold at least its group's age"),
"service_room_key": (True, "§5.5: a proof of a room key makes the room's own service only by a room key"),
"service_proof": (True, "§5.5: a proof of a room key passes live-rooms §10.5.4, or the session is refused"),
"service_no_grant": (True, "§5.5: a proof of a room key names no grant"),
# §4.1, §4.2
"issuers_count": (True, "1 to 16 issuers"),
"issuer_name": (False, "an issuer name is 1 to 255 bytes of ! to ~"),
"policy_issuer_order": (True, "issuers in strictly ascending order of name"),
"token_type_2": (True, "an issuer's token_type is 2"),
"keys_count": (False, "1 to 8 keys an issuer"),
"policy_key_order": (True, "an issuer's keys in strictly ascending order of threshold"),
"threshold_range": (False, "a threshold is 1 to 21"),
"policy_key_twice": (True, "no key twice in a policy"),
"spki_algorithm": (True, "a key's AlgorithmIdentifier is RFC 9578 A.2's"),
"spki_modulus": (True, "a key's modulus is of 2048 bits"),
"window_range": (True, "window_s is 60 to 86 400"),
"policy_room": (True, "a policy names the room"),
"policy_signer": (True, "a policy is signed by the room authority"),
# §5.2
"token_length": (True, "step 1: a token is 354 bytes"),
"token_type": (True, "step 1: a token is of type 0x0002"),
"challenge_digest": (True, "step 2: the challenge digest is of a challenge sent"),
"challenge_form": (True, "step 2: the challenge is of type 0x0002 with a context of 0 or 32 bytes"),
"scope_grammar": (True, "step 2: over MOQT, origin_info reads as a MoQAuthorizationInfo (live-rooms §10.4)"),
"scope_room": (True, "step 2: over MOQT, origin_info is the room's one scope, byte for byte"),
"scope_action_table": (True, "step 2: a scope's actions are of the draft's table"),
"origin_names": (True, "step 2: over HTTP, origin_info is a list of server names"),
"origin_listed": (True, "step 2: over HTTP, the list holds the challenger's own name"),
"issuer_match": (True, "step 3: the key is of the issuer the challenge names"),
"pss": (True, "step 4: the authenticator verifies"),
"replay": (True, "step 5: no nonce redeemed twice"),
"threshold": (True, "step 6: the key's threshold is at least the age asked"),
# §6
"device_challenge": (True, "a credential is issued for the device challenge"),
"anchor_issuer": (True, "a credential's key is one the anchor lists for that issuer"),
"anchor_key": (False, "a credential's key id is of the anchor's key it is checked under"),
"credential_pss": (True, "a credential's authenticator verifies"),
"predicate_verdict": (True, "an object is presented only at a state at least its age"),
# §7
"identifies_top": (True, "a member added at a receipt's top is receipt_identifies"),
"identifies_cell": (True, "a member added in a cell is receipt_identifies"),
# A receipt whose start is not below its end, or whose window_s is 0, also has every cell
# outside its span; only one that holds no cell tells these two rules from cell_overlap.
"receipt_span": (False, "start is below end, apart from the cell-overlap rule"),
"window_zero": (False, "window_s is not 0, apart from the cell-overlap rule"),
"cells_order": (True, "cells in ascending order of window, issuer, threshold"),
"cell_overlap": (True, "a cell's window overlaps the span"),
"suppressed_flag": (False, "suppressed is 0 or 1"),
"suppressed_count": (True, "a suppressed cell states no count"),
"receipt_room": (True, "a receipt names the room"),
"receipt_signer": (True, "a receipt is signed by a room key"),
"receipt_policy": (True, "a receipt names the policy's link hash"),
"receipt_window": (True, "a receipt's window_s is the policy's"),
"receipt_issuer": (True, "a cell's issuer and threshold are a key of the policy"),
"count_not_suppressed": (True, "no cell under 50 stated unsuppressed"),
"admission_span": (True, "only admissions in the span are counted"),
"suppress_small": (True, "§7.2 suppresses a cell under 50"),
"count_mismatch": (True, "the cells are those the admissions write"),
# §5.4, §8
"room_verifier": (True, "the room verifier's refusal is the case's"),
"age_form": (True, "a grant's age reads"),
"minor_age": (True, "a minor's grant states no age of 18 or more"),
"profile_mode": (True, "rule 1: the recording policy's mode is 0"),
"profile_record": (True, "rule 2: no role holds the record power"),
"profile_seek": (True, "rule 2: no role holds the seek power"),
"profile_chat": (True, "rule 3: no role that publishes nothing permits chat"),
"profile_reaction": (True, "rule 3: no role that publishes nothing permits reaction"),
"rule4_chat": (True, "rule 4: no grant that publishes nothing permits chat"),
"rule4_reaction": (False, "rule 4: no grant that publishes nothing permits reaction"),
"exchange_new": (True, "a grant_exchange puts its new grant into effect"),
"minor_profile": (True, "a minor's publishing grant takes effect only in a room meeting the profile"),
}
DROPPED = set()
def on(check):
return check not in DROPPED
class Refuse(Exception):
def __init__(self, reason, at=None):
super().__init__(reason)
self.reason, self.at = reason, at
class Bad(Exception):
"""A record that does not read: `record_malformed`."""
def need(cond, check=None):
if not cond and (check is None or on(check)):
raise Bad(check or "a member breaks its rule")
def sha256(b):
return hashlib.sha256(b).digest()
# ---------------------------------------------------------------- §2 reading a record
class Disallowed:
"""A value RFC 8259 admits and jwp-receipts §2.4.3 does not: a boolean, or a number that is
not an unsigned integer of at most 2^53 - 1."""
def __init__(self, what):
self.what = what
_PLAIN = re.compile(r'[^"\\\x00-\x1f\ud800-\udfff]+')
_NUMBER = re.compile(r"-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?(?:[eE][+-]?[0-9]+)?")
_HEX4 = re.compile(r"[0-9A-Fa-f]{4}")
_ESCAPES = {'"': '"', "\\": "\\", "/": "/", "b": "\b", "f": "\f", "n": "\n", "r": "\r", "t": "\t"}
def read_json(text):
"""jwp-receipts §3.1 step 1: RFC 8259's grammar (no byte-order mark, no unpaired
surrogate), no member named twice, at most 32 levels, the record itself the first; an
object. A boolean or a number past §2.4.3's is read as Disallowed, for `encodings`."""
if not isinstance(text, str):
raise Bad("not a text")
n = len(text)
def ws(i):
while i < n and text[i] in " \t\n\r":
i += 1
return i
def string(i):
i += 1
out = []
while True:
m = _PLAIN.match(text, i)
if m:
out.append(m.group())
i = m.end()
if i >= n:
raise Bad("an unterminated string")
c = text[i]
if c == '"':
return "".join(out), i + 1
if c != "\\":
raise Bad("a control character or an unpaired surrogate in a string")
e = text[i + 1] if i + 1 < n else ""
if e in _ESCAPES:
out.append(_ESCAPES[e])
i += 2
continue
if e != "u" or not _HEX4.fullmatch(text, i + 2, i + 6):
raise Bad("an escape RFC 8259 does not have")
u = int(text[i + 2:i + 6], 16)
i += 6
if 0xD800 <= u <= 0xDBFF:
if text.startswith("\\u", i) and _HEX4.fullmatch(text, i + 2, i + 6):
lo = int(text[i + 2:i + 6], 16)
if 0xDC00 <= lo <= 0xDFFF:
out.append(chr(0x10000 + ((u - 0xD800) << 10) + (lo - 0xDC00)))
i += 6
continue
raise Bad("an unpaired surrogate")
if 0xDC00 <= u <= 0xDFFF:
raise Bad("an unpaired surrogate")
out.append(chr(u))
def value(i, depth):
i = ws(i)
if i >= n:
raise Bad("no value")
c = text[i]
if c in "{[":
if depth > MAX_DEPTH and on("json_depth"):
raise Bad("nested past 32 levels")
obj, close = ({}, "}") if c == "{" else ([], "]")
i = ws(i + 1)
if i < n and text[i] == close:
return obj, i + 1
while True:
if c == "{":
i = ws(i)
if i >= n or text[i] != '"':
raise Bad("a member name that is not a string")
k, i = string(i)
i = ws(i)
if i >= n or text[i] != ":":
raise Bad("a member without a colon")
v, i = value(i + 1, depth + 1)
if k in obj and on("json_duplicate"):
raise Bad("a member named twice")
obj[k] = v
else:
v, i = value(i, depth + 1)
obj.append(v)
i = ws(i)
if i < n and text[i] == ",":
i += 1
continue
if i < n and text[i] == close:
return obj, i + 1
raise Bad("an unterminated container")
if c == '"':
return string(i)
m = _NUMBER.match(text, i)
if m:
s = m.group()
if s[0] == "-" or "." in s or "e" in s or "E" in s:
if not on("json_encodings") and float(s).is_integer() and float(s) >= 0:
return int(float(s)), m.end() # a weakened reader: 1.0 read as 1
return Disallowed("a number that is not an unsigned integer"), m.end()
if len(s) > 16 or int(s) > MAX_INT:
return Disallowed("an integer past 2^53 - 1"), m.end()
return int(s), m.end()
lax = not on("json_encodings") # a weakened reader: true read as 1
for word, v in (("null", None), ("true", 1 if lax else Disallowed("a boolean")),
("false", 0 if lax else Disallowed("a boolean"))):
if text.startswith(word, i):
return v, i + len(word)
raise Bad("not JSON")
try:
v, i = value(0, 1)
except RecursionError:
raise Bad("nested too deep to read")
if ws(i) != n:
raise Bad("text after the value")
if not isinstance(v, dict):
raise Bad("a record that is not an object")
return v
def encodings(v):
"""jwp-receipts §2.4.3: no boolean, no number but an unsigned integer of at most 2^53 - 1."""
if isinstance(v, Disallowed):
need(False, "json_encodings")
elif isinstance(v, dict):
for x in v.values():
encodings(x)
elif isinstance(v, list):
for x in v:
encodings(x)
def jstr(s):
out = ['"']
for c in s:
o = ord(c)
if c == '"':
out.append('\\"')
elif c == "\\":
out.append("\\\\")
elif c == "\b":
out.append("\\b")
elif c == "\f":
out.append("\\f")
elif c == "\n":
out.append("\\n")
elif c == "\r":
out.append("\\r")
elif c == "\t":
out.append("\\t")
elif o < 0x20:
out.append("\\u%04x" % o)
else:
out.append(c)
out.append('"')
return "".join(out)
def jcs(v):
"""RFC 8785 on the values a record carries (jwp-receipts §2.4.3): names sorted by UTF-16
code units, no whitespace, the short escapes, integers in plain decimal."""
if v is None:
return "null"
if isinstance(v, bool) or isinstance(v, Disallowed):
raise Bad("a value JCS is not given here")
if isinstance(v, int):
return str(v)
if isinstance(v, str):
return jstr(v)
if isinstance(v, list):
return "[" + ",".join(jcs(x) for x in v) + "]"
keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
return "{" + ",".join(jstr(k) + ":" + jcs(v[k]) for k in keys) + "}"
_LINKS = {}
def link_hash(v):
"""§2: BLAKE3 over the link domain and the record's JCS."""
b = LINK_DOMAIN + jcs(v).encode("utf-8")
if b not in _LINKS:
_LINKS[b] = blake3(b)
return _LINKS[b]
def hexb(v, n, check=None):
need(isinstance(v, str) and HEX.fullmatch(v) is not None)
need(len(v) == 2 * n, check)
return bytes.fromhex(v)
def opt_hex(v, n):
return None if v is None else hexb(v, n)
def integer(v):
need(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT)
return v
def term(v):
"""jwp-receipts §2.4.3."""
need(isinstance(v, str) and 0 < len(v.encode("utf-8")) <= 2048
and not any(c in WHITE_SPACE or ord(c) < 0x20 or 0x7F <= ord(c) <= 0x9F for c in v))
return v
def absolute_uri(v):
"""jwp-receipts §2.4.3: a term with a scheme (RFC 3986 §3.1), a colon and a non-empty
remainder."""
need(isinstance(v, str))
if on("absolute_uri"):
term(v)
need(URI_SCHEME.fullmatch(v) is not None)
return v
def printable(v, most):
"""1 to `most` bytes of ! to ~."""
return isinstance(v, str) and 1 <= len(v) <= most and all("!" <= c <= "~" for c in v)
def date_time(s):
"""jwp-receipts §2.4.3: RFC 3339 §5.6's date-time, with §5.7's days."""
m = DATE_TIME.fullmatch(s) if isinstance(s, str) else None
need(m is not None, "date_time")
if m is None:
return
y, mo, d, h, mi, sec = (int(m.group(i)) for i in range(1, 7))
leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
need(1 <= mo <= 12 and 1 <= d <= days[mo - 1] and h <= 23 and mi <= 59 and sec <= 60, "date_time")
if m.group(9) is not None:
need(int(m.group(9)) <= 23 and int(m.group(10)) <= 59, "date_time")
def members(v, names):
need(isinstance(v, dict))
need(set(v) == set(names), "exact_members")
return v
def read_record(text, kind, names, fields, receipt=False):
"""§2 (with §7.3's change for a receipt): the record, or Refuse. Its signature is checked
by `verify_sig`, after what each record's section checks first."""
try:
v = read_json(text)
except Bad:
raise Refuse("record_malformed")
if receipt:
identifies(v)
try:
encodings(v)
members(v, names)
need(v["kind"] == kind, "kind")
r = {"value": v, "signer": hexb(v["signer_pubkey"], 32), "sig": hexb(v["sig"], 64)}
date_time(v["issued_at"])
r.update(fields(v))
except (Bad, KeyError, TypeError, AttributeError, ValueError):
raise Refuse("record_malformed")
return r
def verify_sig(r):
"""§2: Ed25519 over the signature domain and the JCS of the record without `sig`, strictly
(SPEC §7): `bad_key` or `bad_signature`."""
if not on("signature"):
return
unsigned = {k: x for k, x in r["value"].items() if k != "sig"}
skip = () if on("strict_key") else ("small-order-key",)
bad = ED.verify(r["signer"], SIGN_DOMAIN + jcs(unsigned).encode("utf-8"), r["sig"], skip)
if bad:
raise Refuse(bad)
# ---------------------------------------------------------------- §3 rating claims
def subject_of(s):
"""§3.1's four forms, and §3.2."""
need(isinstance(s, dict))
t = s.get("type")
if t == "object":
members(s, ["type", "content_hash"])
return {"type": t, "hash": hexb(s["content_hash"], 32, "subject_size")}
if t == "track":
members(s, ["type", "track", "from_group"])
return {"type": t, "hash": hexb(s["track"], 32, "subject_size"), "from": integer(s["from_group"]), "to": None}
if t == "span":
members(s, ["type", "track", "from_group", "to_group"])
a, z = integer(s["from_group"]), integer(s["to_group"])
need(z >= a, "span_order")
return {"type": t, "hash": hexb(s["track"], 32, "subject_size"), "from": a, "to": z}
if t == "room":
members(s, ["type", "room"])
return {"type": t, "hash": hexb(s["room"], 32, "subject_size")}
raise Bad("a subject of none of the four forms")
def claim_fields(v):
"""§3.1, §3.2."""
c = {"subject": subject_of(v["subject"])}
rs = v["ratings"]
need(isinstance(rs, list))
need(1 <= len(rs) <= 8, "ratings_count")
c["ratings"] = []
for x in rs:
members(x, ["scheme", "value", "min_age"])
scheme = absolute_uri(x["scheme"])
need(isinstance(x["value"], str))
need(printable(x["value"], 64), "value_bytes")
age = integer(x["min_age"])
need(age <= MAX_AGE, "min_age")
c["ratings"].append((scheme, x["value"], age))
schemes = [r[0].encode("utf-8") for r in c["ratings"]]
need(all(a < b for a, b in zip(schemes, schemes[1:])), "ratings_order")
ins = v["inputs"]
need(isinstance(ins, list))
need(len(ins) <= 16, "inputs_count")
c["inputs"] = []
for x in ins:
members(x, ["content_hash", "rating"])
c["inputs"].append({"content_hash": hexb(x["content_hash"], 32), "rating": opt_hex(x["rating"], 32)})
hashes = [x["content_hash"] for x in c["inputs"]]
need(len(set(hashes)) == len(hashes), "inputs_distinct")
c["record"] = opt_hex(v["record"], 32)
c["age"] = max((r[2] for r in c["ratings"]), default=0) # §3.2: the claim's age
return c
def read_claim(text):
c = read_record(text, "wai-rating", CLAIM, claim_fields)
verify_sig(c)
c["link"] = link_hash(c["value"])
return c
def consistent(claims):
"""§3.2: one scheme's one value means one age; the later claim is refused."""
said = {}
for i, c in enumerate(claims):
for scheme, value, age in c["ratings"]:
first = said.setdefault((scheme, value), age)
if first != age and on("consistency"):
raise Refuse("rating_inconsistent", ["claim", i])
def carries(c, claims, by_link):
"""§3.4: for each input, rules 1 to 3 in their order."""
have = {}
for scheme, _, age in c["ratings"]:
have[scheme] = age
for x in c["inputs"]:
named = None
if x["rating"] is not None: # rule 1
named = by_link.get(x["rating"])
if named is None and on("rule1_held"):
raise Refuse("rating_unknown")
if (named is not None and named["subject"]["type"] == "object" and named["subject"]["hash"] != x["content_hash"]
and on("rule1_subject")):
raise Refuse("rating_unknown")
covering = [h for h in claims if h["subject"]["type"] == "object" and h["subject"]["hash"] == x["content_hash"]]
if x["rating"] is None and covering and on("rule2"): # rule 2
raise Refuse("rating_dropped")
for h in covering + ([named] if named is not None else []): # rule 3
for scheme, _, age in h["ratings"]:
if scheme not in have:
if on("rule3_scheme"):
raise Refuse("rating_dropped")
elif have[scheme] < age and on("rule3_age"):
raise Refuse("rating_dropped")
def covers(c, content_hash, track, group, room):
"""§3.3: whether claim c covers an object of that content hash, held at (track, group) of
the room `room` (any of them None where unknown)."""
s = c["subject"]
if s["type"] == "object":
return content_hash is not None and s["hash"] == content_hash
if s["type"] in ("track", "span"):
if track is None or group is None or s["hash"] != track:
return False
if s["from"] > group and on("track_from"):
return False
return s["type"] == "track" or group <= s["to"] or not on("span_end")
return room is not None and s["hash"] == room
def age_under(claims, content_hash, track, group, room):
"""§3.3: the greatest age of the claims that cover the object, 0 when none does."""
return max((c["age"] for c in claims if covers(c, content_hash, track, group, room)), default=0)
def render(field):
"""MOQT §8.8: a-z, A-Z, 0-9 and _ as themselves; any other byte as . and two hex digits."""
return "".join(chr(b) if (0x30 <= b <= 0x39 or 0x41 <= b <= 0x5A or 0x61 <= b <= 0x7A or b == 0x5F) else ".%02x" % b
for b in field.encode("utf-8"))
def track_digest(namespace, name):
"""§2, live-rooms §6.2: SHA-256 over the rendering of the namespace, a zero byte, the name."""
return sha256("-".join(render(f) for f in namespace).encode("ascii") + b"\x00" + name.encode("utf-8"))
def honoured(c, d, track, owner, carried=()):
"""§3.6: True for a claim a relay of the room honours, False for one it cannot attribute to
the room, a track it serves (here, the case's track) or an object it carries (`carried`),
which it neither honours nor refuses and which covers nothing it serves; or Refuse. A room
claim is checked for its room first, then for its signer."""
room_keys = (d["authority"], d["clock"])
s = c["subject"]
if s["type"] == "room":
if s["hash"] != d["room"] and on("room_claim_room"):
raise Refuse("wrong_room")
if c["signer"] not in room_keys and on("honour_room_signer"):
raise Refuse("rating_not_authorised")
return True
attributed = s["hash"] == track if s["type"] in ("track", "span") else s["hash"] in carried
if not attributed and on("attribution"):
return False
allowed = set(room_keys) | ({owner} if owner is not None else set())
if c["signer"] not in allowed and on("honour_track_signer"):
raise Refuse("rating_not_authorised")
return True
# ---------------------------------------------------------------- §4 the age policy, RFC 9578 §6.5 keys
def der(b, i, tag):
"""One DER element of `tag` at i (X.690 §10.1: the definite length, in its fewest bytes):
its contents and the index after it."""
need(i + 2 <= len(b) and b[i] == tag)
n = b[i + 1]
i += 2
if n & 0x80:
k = n & 0x7F
need(1 <= k <= 4 and i + k <= len(b) and b[i] != 0)
n = int.from_bytes(b[i:i + k], "big")
need(n >= 0x80)
i += k
need(i + n <= len(b))
return b[i:i + n], i + n
def der_positive(c):
"""A DER INTEGER's contents, read as a positive integer in its fewest bytes."""
need(len(c) >= 1 and not (len(c) > 1 and c[0] == 0 and c[1] < 0x80) and c[0] < 0x80)
n = int.from_bytes(c, "big")
need(n > 0)
return n
def read_spki(b):
"""§4.2, RFC 9578 §6.5: the DER SubjectPublicKeyInfo of RFC 9578 A.2's AlgorithmIdentifier,
byte for byte, whose key is an RSAPublicKey of a 2048-bit modulus and a positive exponent:
(n, e)."""
body, end = der(b, 0, 0x30)
need(end == len(b))
_, j = der(body, 0, 0x30)
need(body[:j] == PSS_ALGORITHM, "spki_algorithm")
bits, end = der(body, j, 0x03)
need(end == len(body) and len(bits) >= 1 and bits[0] == 0)
rsa, end = der(bits, 1, 0x30)
need(end == len(bits))
nc, k = der(rsa, 0, 0x02)
ec, k = der(rsa, k, 0x02)
need(k == len(rsa))
n, e = der_positive(nc), der_positive(ec)
need(n.bit_length() == 2048, "spki_modulus")
return n, e
def spki_or_none(b):
try:
return read_spki(b)
except Bad:
return None
def policy_fields(v):
"""§4.1, §4.2."""
p = {"room": hexb(v["room"], 32)}
iss = v["issuers"]
need(isinstance(iss, list))
need(1 <= len(iss) <= 16, "issuers_count")
p["issuers"], seen = [], set()
for x in iss:
members(x, ["name", "token_type", "keys"])
need(isinstance(x["name"], str))
need(printable(x["name"], 255), "issuer_name")
need(integer(x["token_type"]) == 2, "token_type_2")
ks = x["keys"]
need(isinstance(ks, list))
need(1 <= len(ks) <= 8, "keys_count")
keys = []
for k in ks:
members(k, ["threshold", "key"])
t = integer(k["threshold"])
need(1 <= t <= MAX_AGE, "threshold_range")
need(isinstance(k["key"], str) and HEX.fullmatch(k["key"]) is not None)
der_bytes = bytes.fromhex(k["key"])
need(der_bytes not in seen, "policy_key_twice")
seen.add(der_bytes)
n, e = read_spki(der_bytes)
keys.append({"threshold": t, "der": der_bytes, "id": sha256(der_bytes), "n": n, "e": e})
need(all(a["threshold"] < b["threshold"] for a, b in zip(keys, keys[1:])), "policy_key_order")
p["issuers"].append({"name": x["name"], "keys": keys})
names = [x["name"].encode("utf-8") for x in p["issuers"]]
need(all(a < b for a, b in zip(names, names[1:])), "policy_issuer_order")
w = integer(v["window_s"])
need(60 <= w <= 86400, "window_range")
p["window_s"] = w
return p
def read_policy(text, d):
"""§4.2, in its order: the policy, or Refuse."""
p = read_record(text, "wai-age-policy", POLICY, policy_fields)
if p["room"] != d["room"] and on("policy_room"):
raise Refuse("wrong_room")
if p["signer"] != d["authority"] and on("policy_signer"):
raise Refuse("signer_not_permitted")
verify_sig(p)
p["link"] = link_hash(p["value"])
return p
# ---------------------------------------------------------------- RFC 8017 RSASSA-PSS, RFC 9577 tokens
def sha384(b):
return hashlib.sha384(b).digest()
def mgf1(seed, length):
"""RFC 8017 Appendix B.2.1, with SHA-384."""
out, counter = b"", 0
while len(out) < length:
out += sha384(seed + counter.to_bytes(4, "big"))
counter += 1
return out[:length]
def emsa_pss_verify(message, em, em_bits):
"""RFC 8017 §9.1.2, with SHA-384 and a salt of 48 bytes: consistent or not."""
h_len, em_len = 48, len(em)
m_hash = sha384(message) # step 2
if em_len < h_len + SALT_LEN + 2: # step 3
return False
if em[-1] != 0xBC: # step 4
return False
masked_db, h = em[:em_len - h_len - 1], em[em_len - h_len - 1:-1] # step 5
zero_bits = 8 * em_len - em_bits
if zero_bits and masked_db[0] >> (8 - zero_bits): # step 6
return False
db = bytes(a ^ b for a, b in zip(masked_db, mgf1(h, em_len - h_len - 1))) # steps 7, 8
db = bytes([db[0] & (0xFF >> zero_bits)]) + db[1:] # step 9
ps_len = em_len - h_len - SALT_LEN - 2
if any(db[:ps_len]) or db[ps_len] != 0x01: # step 10
return False
salt = db[-SALT_LEN:] # step 11
return sha384(b"\x00" * 8 + m_hash + salt) == h # steps 12 to 14
def rsassa_pss_verify(key, message, signature):
"""RFC 8017 §8.1.2 under (n, e)."""
n, e = key
mod_bits = n.bit_length()
if len(signature) != (mod_bits + 7) // 8: # step 1
return False
s = int.from_bytes(signature, "big") # step 2a
if s >= n: # step 2b (RSAVP1)
return False
m = pow(s, e, n)
em_len = (mod_bits - 1 + 7) // 8
if m >= 1 << (8 * em_len): # step 2c (I2OSP)
return False
return emsa_pss_verify(message, m.to_bytes(em_len, "big"), mod_bits - 1) # step 3
def read_challenge(b):
"""§5.2 step 2: the structure of RFC 9577 §2.1.1's TokenChallenge, every field read as bytes,
an issuer_name of at least one byte and no byte after origin_info, which is read by its
binding apart (scope_check): (token_type, issuer_name, redemption_context, origin_info), or
None."""
try:
if len(b) < 4:
return None
token_type, nl = int.from_bytes(b[0:2], "big"), int.from_bytes(b[2:4], "big")
i = 4 + nl
name = b[4:i]
cl = b[i]
ctx = b[i + 1:i + 1 + cl]
i += 1 + cl
ol = int.from_bytes(b[i:i + 2], "big")
origin = b[i + 2:i + 2 + ol]
if nl < 1 or len(name) != nl or len(ctx) != cl or i + 2 > len(b) or len(origin) != ol or i + 2 + ol != len(b):
return None
return token_type, name, ctx, origin
except IndexError:
return None
# live-rooms §10.4: the room's one scope, SUBSCRIBE_NAMESPACE (3), SUBSCRIBE (4), FETCH (7) and
# TRACK_STATUS (8) on MATCH_PREFIX (1) of the tuple of the room id and on the empty track name.
SCOPE_ACTIONS = bytes([3, 4, 7, 8])
MATCH_PREFIX = 1
def room_scope(room):
"""live-rooms §10.4's bytes for the room id `room` (32 bytes): the scopes' length, then the
actions, the namespace match (its type, the tuple, the element) and the track match."""
element = room.hex().encode("ascii")
tuple_ = len(element).to_bytes(2, "big") + element
scope = (bytes([len(SCOPE_ACTIONS)]) + SCOPE_ACTIONS + bytes([MATCH_PREFIX]) + len(tuple_).to_bytes(2, "big") + tuple_
+ bytes([MATCH_PREFIX]) + b"\x00\x00")
return bytes([len(scope)]) + scope
def scope_reads(b):
"""draft-ietf-moq-privacy-pass-auth-03 §3.2.4 in the TLS presentation language: the scopes'
length 1 to 255 and exactly that many bytes of whole MoQAuthScopes; actions 1 to 255 bytes,
each 0 to 8 (§3.2.2); match types 0 to 3 (§3.2.3); every tuple, element and value whole."""
if len(b) < 1 or b[0] == 0 or len(b) != 1 + b[0]:
return False
i, end = 1, len(b)
def u16(at):
if at + 2 > end:
raise IndexError
return int.from_bytes(b[at:at + 2], "big")
try:
while i < end:
a = b[i]
if a == 0 or i + 1 + a > end:
return False
if on("scope_action_table") and any(x > 8 for x in b[i + 1:i + 1 + a]):
return False
i += 1 + a
if b[i] > 3:
return False
n = u16(i + 1)
j, stop = i + 3, i + 3 + n
if stop > end:
return False
while j < stop:
el = u16(j)
j += 2 + el
if j != stop:
return False
i = stop
if b[i] > 3:
return False
i += 3 + u16(i + 1)
if i > end:
return False
return True
except IndexError:
return False
def server_names(b):
"""RFC 9577 §2.1.1: origin_info over HTTP, empty or server names joined by ',' without
whitespace, each of '!' to '~' and no userinfo (§2.1.1.1): the names, or None."""
if not b:
return []
names = b.split(b",")
if any(not n or any(c < 0x21 or c > 0x7E or c == 0x40 for c in n) for n in names):
return None
return names
def scope_check(origin_info, origin, room):
"""§5.2 step 2's origin_info: over MOQT (origin None), live-rooms §10.4's scope of the room;
over HTTP, a list of server names holding the challenger's own."""
if origin is None:
if origin_info == room_scope(room) or not on("scope_room"):
return
if origin_info and not scope_reads(origin_info) and on("scope_grammar"):
raise Refuse("scope_malformed")
raise Refuse("scope_mismatch")
names = server_names(origin_info)
if names is None and on("origin_names"):
raise Refuse("scope_malformed")
if on("origin_listed") and not any(n.lower() == origin.encode("ascii").lower() for n in names or []):
raise Refuse("scope_mismatch")
def device_challenge(name):
"""§6.1: token_type 0x0002, the issuer's name, an empty redemption_context, an empty
origin_info."""
b = name.encode("utf-8")
return TOKEN_TYPE + len(b).to_bytes(2, "big") + b + b"\x00" + b"\x00\x00"
def token_reads(tok):
"""§5.2 step 1."""
return (len(tok) == TOKEN_LEN or not on("token_length")) and (tok[:2] == TOKEN_TYPE or not on("token_type"))
def check_token(policy, challenges, redeemed, tok, required, origin=None):
"""§5.2's six steps, in their order (step 6 only where an age is asked), over MOQT or, with
`origin`, over HTTP: (issuer, threshold, key id), or Refuse. A nonce that passes step 5 is
added to `redeemed`."""
if not token_reads(tok) or len(tok) < SIGNED_LEN: # 1
raise Refuse("token_malformed")
nonce, digest, key_id, authenticator = tok[2:34], tok[34:66], tok[66:98], tok[98:]
sent = [c for c in challenges if sha256(c) == digest or not on("challenge_digest")] # 2
if not sent:
raise Refuse("token_wrong_challenge")
ch = read_challenge(sent[0])
if on("challenge_form") and (ch is None or ch[0] != 2 or len(ch[2]) not in (0, 32)):
raise Refuse("token_wrong_challenge")
if ch is not None:
scope_check(ch[3], origin, policy["room"])
name = ch[1] if ch is not None else b""
issuers = [x for x in policy["issuers"] if x["name"].encode("utf-8") == name or not on("issuer_match")] # 3
found = next(((x, k) for x in issuers for k in x["keys"] if k["id"] == key_id), None)
if found is None:
raise Refuse("issuer_unknown")
issuer, key = found
if on("pss") and not rsassa_pss_verify((key["n"], key["e"]), tok[:SIGNED_LEN], authenticator): # 4
raise Refuse("token_bad_signature")
if nonce in redeemed and on("replay"): # 5
raise Refuse("token_replayed")
redeemed.add(nonce)
if required is not None and key["threshold"] < required and on("threshold"): # 6
raise Refuse("below_threshold")
return issuer["name"], key["threshold"], key_id
def holds(tok, anchor):
"""§6.1: the threshold of the anchor's key a credential holds under, or None."""
if len(tok) != TOKEN_LEN or tok[:2] != TOKEN_TYPE:
return None
digest, key_id, authenticator = tok[34:66], tok[66:98], tok[98:]
best = None
for issuer in sorted({a["issuer"] for a in anchor}):
if sha256(device_challenge(issuer)) != digest and on("device_challenge"):
continue
for a in anchor:
if (a["issuer"] != issuer and on("anchor_issuer")) or (sha256(a["key"]) != key_id and on("anchor_key")):
continue
key = spki_or_none(a["key"])
if key is None or (on("credential_pss") and not rsassa_pss_verify(key, tok[:SIGNED_LEN], authenticator)):
continue
best = a["threshold"] if best is None else max(best, a["threshold"])
return best
# ---------------------------------------------------------------- §7 the admission receipt
def identifies(v):
"""§7.3: a member §7.1 does not list, at the top or in a cell, before anything else."""
extra = on("identifies_top") and bool(set(v) - set(RECEIPT))
cells = v.get("cells")
if isinstance(cells, list) and on("identifies_cell"):
extra = extra or any(isinstance(c, dict) and bool(set(c) - set(CELL)) for c in cells)
if extra:
raise Refuse("receipt_identifies")
def overlaps(w, window_s, start, end):
"""§7.1: window w is [w W, (w + 1) W) in room-clock microseconds, W = window_s 10^6."""
size = window_s * 1000000
return w * size < end and (w + 1) * size > start
def receipt_fields(v):
"""§7.1, §7.3's rules of form."""
r = {"room": hexb(v["room"], 32), "policy": hexb(v["policy"], 32), "start": integer(v["start"]),
"end": integer(v["end"]), "window_s": integer(v["window_s"])}
need(r["start"] < r["end"], "receipt_span")
need(r["window_s"] != 0, "window_zero")
need(isinstance(v["cells"], list))
r["cells"] = []
for c in v["cells"]:
members(c, CELL)
need(isinstance(c["issuer"], str))
cell = (integer(c["window"]), c["issuer"], integer(c["threshold"]), integer(c["admitted"]), integer(c["suppressed"]))
need(overlaps(cell[0], r["window_s"], r["start"], r["end"]), "cell_overlap")
need(cell[4] in (0, 1), "suppressed_flag")
need(cell[4] == 0 or cell[3] == 0, "suppressed_count")
r["cells"].append(cell)
order = [(c[0], c[1].encode("utf-8"), c[2]) for c in r["cells"]]
need(all(a < b for a, b in zip(order, order[1:])), "cells_order")
return r
def read_receipt(text, d, policy):
"""§7.3, in its order: the receipt, or Refuse."""
r = read_record(text, "wai-age-admissions", RECEIPT, receipt_fields, receipt=True)
if r["room"] != d["room"] and on("receipt_room"):
raise Refuse("wrong_room")
if r["signer"] not in (d["authority"], d["clock"]) and on("receipt_signer"):
raise Refuse("signer_not_permitted")
verify_sig(r)
keys = {(x["name"], k["threshold"]) for x in policy["issuers"] for k in x["keys"]}
if ((r["policy"] != policy["link"] and on("receipt_policy"))
or (r["window_s"] != policy["window_s"] and on("receipt_window"))
or (on("receipt_issuer") and any((c[1], c[2]) not in keys for c in r["cells"]))):
raise Refuse("receipt_mismatch")
if any(c[4] == 0 and c[3] < FLOOR for c in r["cells"]) and on("count_not_suppressed"):
raise Refuse("count_not_suppressed")
r["link"] = link_hash(r["value"])
return r
def write_cells(admitted, start, end, window_s):
"""§7.2: the cells of the admissions in the span, in §7.3's order."""
counts = {}
for t, issuer, threshold in admitted:
if start <= t < end or not on("admission_span"):
k = (t // (window_s * 1000000), issuer, threshold)
counts[k] = counts.get(k, 0) + 1
cells = []
for k in sorted(counts, key=lambda k: (k[0], k[1].encode("utf-8"), k[2])):
n = counts[k]
small = n < FLOOR and on("suppress_small")
cells.append(k + ((0, 1) if small else (n, 0)))
return cells
def cell_dict(c):
return {"window": c[0], "issuer": c[1], "threshold": c[2], "admitted": c[3], "suppressed": c[4]}
# ---------------------------------------------------------------- §5.4, §8 grants and the profile
def grant_age(attrs):
"""§5.4: the threshold a grant states, or Refuse("age_attr_invalid")."""
v = attrs.get("age")
if v is None:
age = 0
else:
if on("age_form") and not (AGE_ATTR.fullmatch(v) and 1 <= int(v) <= MAX_AGE):
raise Refuse("age_attr_invalid")
age = int(v)
if attrs.get("minor") == "true" and age >= 18 and on("minor_age"):
raise Refuse("age_attr_invalid")
return age
def breaks_rule_4(g):
"""§8.1 rule 4: a grant that publishes nothing and permits chat or reaction in its policy."""
if g["publish"] != 0:
return False
return ((LEV.permits(g["policy"], CHAT) and on("rule4_chat"))
or (LEV.permits(g["policy"], REACTION) and on("rule4_reaction")))
def meets_profile(d):
"""§8.1 rules 1 to 3, the descriptor's."""
if d["mode"] != 0 and on("profile_mode"):
return False
for role in d["roles"]:
if role["powers"] & RECORD_POWER and on("profile_record"):
return False
if role["powers"] & SEEK_POWER and on("profile_seek"):
return False
if role["publish"] == 0:
if LRV.has(role["events"], CHAT) and on("profile_chat"):
return False
if LRV.has(role["events"], REACTION) and on("profile_reaction"):
return False
return True
# ---------------------------------------------------------------- §9.1 the cases
def descriptor_of(case):
return LRV.read_descriptor(bytes.fromhex(case["descriptor"]))
def policy_at(case, d):
try:
return read_policy(case["policy"], d)
except Refuse as e:
raise Refuse(e.reason, ["policy"])
def case_ratings(case, doc):
claims = []
for i, t in enumerate(case["claims"]): # each reads and verifies, in order
try:
claims.append(read_claim(t))
except Refuse as e:
raise Refuse(e.reason, ["claim", i])
consistent(claims) # §3.2
by_link = {c["link"]: c for c in claims}
for i, c in enumerate(claims): # §3.4
try:
carries(c, claims, by_link)
except Refuse as e:
raise Refuse(e.reason, ["claim", i])
return {"claims": [{"link": c["link"].hex(), "signer": c["signer"].hex(), "subject": c["subject"]["type"],
"age": c["age"], "derived": 1 if c["inputs"] else 0} for c in claims]}
def case_admission(case, doc):
d = descriptor_of(case)
policy = policy_at(case, d)
try:
issuer, threshold, key_id = check_token(policy, [bytes.fromhex(c) for c in case["challenges"]],
{bytes.fromhex(n) for n in case["redeemed"]},
bytes.fromhex(case["token"]), case["required"], case["origin"])
except Refuse as e:
raise Refuse(e.reason, ["token"])
return {"issuer": issuer, "threshold": threshold, "key_id": key_id.hex()}
def case_serving(case, doc):
d = descriptor_of(case)
ns, name = case["track"]["namespace"], case["track"]["name"]
track = track_digest(ns, name)
room = bytes.fromhex(ns[0]) if HEX64.fullmatch(ns[0]) else None # live-rooms §6.1
owner = bytes.fromhex(ns[1]) if len(ns) > 1 and HEX64.fullmatch(ns[1]) else None
claims = []
for i, t in enumerate(case["claims"]): # in order: reads, verifies, §3.6
try:
c = read_claim(t)
c["honoured"] = honoured(c, d, track, owner) # a serving case names no object carried
except Refuse as e:
raise Refuse(e.reason, ["claim", i])
claims.append(c)
out = []
for q in case["requests"]:
held = [claims[k] for k in q["held"] if claims[k]["honoured"]]
age = age_under(held, None, track, q["group"], room)
try:
own = "service" in q and room_service(d, q["service"])
except Refuse as e:
out.append({"age": age, "outcome": "refuse", "reason": e.reason})
continue
if own:
out.append({"age": age, "outcome": "serve", "service": True})
elif q["threshold"] >= age or not on("serve_rule"):
out.append({"age": age, "outcome": "serve"})
else:
out.append({"age": age, "outcome": "refuse", "reason": "below_threshold"})
return {"track": track.hex(), "requests": out}
NO_GRANT = bytes(32)
def room_service(d, s):
"""§5.5: the proof of a room key a session presented, at a relay holding the session's
`challenge` and receiving `context`. It is checked as live-rooms §10.5.4 checks a proof,
the grant's digest 32 bytes of zero and the principal the proof's own key: a proof that
does not pass refuses the session with that section's reason. True when it passes by a room
key (the authority's or the clock's): the room's own service, which the serving rule does
not hold; False for any other key, a session served by its threshold."""
proof = bytes.fromhex(s["proof"])
challenge, context = bytes.fromhex(s["challenge"]), s["context"]
try:
p = LRV.read_proof(proof) # step 2
except LEV.Refuse as e:
raise Refuse(e.reason)
if p["grant"] != NO_GRANT and on("service_no_grant"): # step 3
raise Refuse("pop_wrong_grant")
if on("service_proof"):
if ED.verify(p["key"], LRV.PROOF_DOMAIN + p["body"], p["sig"]) is not None: # step 5
raise Refuse("pop_bad_signature")
if p["challenge"] != challenge: # step 6: none consumed
raise Refuse("pop_wrong_challenge")
if p["context"] != context: # step 7
raise Refuse("pop_wrong_context")
return p["key"] in (d["authority"], d["clock"]) or not on("service_room_key")
def case_predicate(case, doc):
anchor = [{"issuer": a["issuer"], "threshold": a["threshold"], "key": bytes.fromhex(a["key"])} for a in case["anchor"]]
state = 0
for h in case["credentials"]:
t = holds(bytes.fromhex(h), anchor)
if t is not None:
state = max(state, t)
claims = []
for t in case["claims"]: # §6.2: every claim that verifies
try:
claims.append(read_claim(t))
except Refuse:
continue
objects = []
for o in case["objects"]:
h = None if o["content_hash"] is None else bytes.fromhex(o["content_hash"])
x = None if o["track"] is None else bytes.fromhex(o["track"])
age = age_under(claims, h, x, o["group"], None)
present = state >= age or not on("predicate_verdict")
objects.append({"age": age, "verdict": "present" if present else "withhold"})
return {"state": state, "objects": objects}
def case_receipt(case, doc):
d = descriptor_of(case)
policy = policy_at(case, d)
admitted = None
if case["admissions"] is not None:
admitted, redeemed = [], set()
for i, a in enumerate(doc["admission_logs"][case["admissions"]]):
try:
issuer, threshold, _ = check_token(policy, [bytes.fromhex(c) for c in a["challenges"]], redeemed,
bytes.fromhex(a["token"]), None)
except Refuse as e:
raise Refuse(e.reason, ["admission", i])
admitted.append((a["t"], issuer, threshold))
try:
r = read_receipt(case["receipt"], d, policy)
if admitted is not None and on("count_mismatch"):
if write_cells(admitted, r["start"], r["end"], policy["window_s"]) != r["cells"]:
raise Refuse("count_mismatch")
except Refuse as e:
raise Refuse(e.reason, ["receipt"])
return {"link": r["link"].hex(), "cells": [cell_dict(c) for c in r["cells"]]}
def case_room(case, doc):
descriptor = bytes.fromhex(case["descriptor"])
grants = [bytes.fromhex(g) for g in case["grants"]]
chains = [[bytes.fromhex(r) for r in c] for c in case["chains"]]
room = LRV.verify_room(descriptor, grants, case["key_log"], chains)
if room["outcome"] == "refuse" and on("room_verifier"):
raise Refuse(room["reason"], ["room"] + list(room["at"]))
d = LRV.read_descriptor(descriptor)
held, _ = LRV.check_grants(d, grants)
meets = meets_profile(d)
thresholds, minor = [], set()
for i, g in enumerate(held):
attrs = dict(LRV.read_attrs(g["attrs"]) or [])
try:
thresholds.append(grant_age(attrs))
except Refuse as e:
raise Refuse(e.reason, ["grant", i])
if g["publish"] != 0 and attrs.get("minor") == "true": # §8.2
minor.add(g["digest"])
took, seen = [], set()
for c, i, _ in room["order"]: # §8.2, in the merged order
ev = LEV.read_record(chains[c][i])
if ev["digest"] in seen:
continue
seen.add(ev["digest"])
if ev["kind"] not in (JOIN, EXCHANGE) or not LRV.is_room_key(d, ev["principal"]):
continue
x = LRV.read_detail(ev["kind"], ev["payload"][32:])
if ev["kind"] == EXCHANGE and not on("exchange_new"):
continue
g = x["grant"] if ev["kind"] == JOIN else x["new"]
if g not in minor:
continue
if not took and on("minor_profile"): # the first that takes effect
if not meets:
raise Refuse("minor_profile_unmet", ["record", c, i])
for k, h in enumerate(held):
if breaks_rule_4(h):
raise Refuse("minor_profile_unmet", ["grant", k])
took.append(g.hex())
return {"room_outcome": room["outcome"], "meets_profile": 1 if meets else 0, "thresholds": thresholds,
"minor_grants": took}
CASES = {"ratings": case_ratings, "admission": case_admission, "serving": case_serving, "predicate": case_predicate,
"receipt": case_receipt, "room": case_room}
def outcome(case, doc):
try:
out = {"outcome": "accept"}
out.update(CASES[case["kind"]](case, doc))
return out
except Refuse as e:
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
def run(case, doc):
try:
return outcome(case, doc)
except Exception as e: # noqa: BLE001 — a malformed copy fails, whatever it breaks
return {"outcome": "unreadable: %s: %s" % (type(e).__name__, e)}
# ---------------------------------------------------------------- the corpus's other facts
def a2_vector(v):
"""README, §12: RFC 9578 Appendix A.2's first vector verifies: the failures."""
fails = []
pk, ch, tok = bytes.fromhex(v["pkI"]), bytes.fromhex(v["token_challenge"]), bytes.fromhex(v["token"])
if len(tok) != TOKEN_LEN or tok[:2] != TOKEN_TYPE:
return ["its token does not read (§5.2 step 1)"]
c = read_challenge(ch)
if c is None or c[0] != 2 or len(c[2]) not in (0, 32):
fails.append("its token_challenge does not read as RFC 9577 §2.1.1's")
if sha256(ch) != tok[34:66]:
fails.append("its challenge_digest is not the SHA-256 of token_challenge")
if sha256(pk) != tok[66:98]:
fails.append("its token_key_id is not the SHA-256 of pkI")
key = spki_or_none(pk)
if key is None:
fails.append("its pkI does not read as §4.2's key")
elif not rsassa_pss_verify(key, tok[:SIGNED_LEN], tok[SIGNED_LEN:]):
fails.append("its authenticator does not verify under pkI")
if pk[4:4 + len(PSS_ALGORITHM)] != PSS_ALGORITHM:
fails.append("its pkI's AlgorithmIdentifier is not the one this verifier pins")
if key is not None and "blind" in v:
# RFC 9578 §6.1 to §6.3, RFC 9474 §4.2 to §4.4: with s the authenticator and r the blind,
# the issuer answers m^d * r = s * r, and the blinded message is m * r^e = (s * r)^e.
n, e = key
s, r = int.from_bytes(tok[SIGNED_LEN:], "big"), int.from_bytes(bytes.fromhex(v["blind"]), "big")
req, resp = bytes.fromhex(v["token_request"]), bytes.fromhex(v["token_response"])
if req[:2] != TOKEN_TYPE or req[2] != tok[97] or req[3:] != pow(s * r % n, e, n).to_bytes(256, "big"):
fails.append("its token_request is not the blinded token input under pkI")
if resp != (s * r % n).to_bytes(256, "big"):
fails.append("its token_response does not unblind to its authenticator")
if tok[2:34] != bytes.fromhex(v["nonce"]):
fails.append("its token's nonce is not the vector's")
return fails
def issuer_key_facts(doc):
"""README: the informative issuer keys each read by §4.2, and the 18 key is RFC 9578
A.2's: the failures."""
fails = []
for k in doc["issuer_keys"]:
if spki_or_none(bytes.fromhex(k["key"])) is None:
fails.append("issuer key %s %d does not read by §4.2" % (k["issuer"], k["threshold"]))
a2 = [k for k in doc["issuer_keys"] if "9578" in k["source"]]
if len(a2) != 1 or a2[0]["key"] != doc["rfc9578_a2_vector1"]["pkI"]:
fails.append("the issuer key from RFC 9578 Appendix A.2 is not the vector's pkI")
return fails
def check(doc):
"""Failures of a corpus document: a case's outcome other than the one stated, or a fact of
the corpus that does not hold."""
fails = ["RFC 9578 A.2 vector 1: " + f for f in a2_vector(doc["rfc9578_a2_vector1"])]
fails += issuer_key_facts(doc)
for c in doc["cases"]:
got = run(c, doc)
if got != c["expect"]:
fails.append("%s: %s" % (c["name"], json.dumps(got)[:300]))
return fails
# ---------------------------------------------------------------- the extension's table
def section(text, head):
at = text.index("\n" + head)
nxt = re.search(r"\n#{2,3} ", text[at + 1:])
return text[at: at + 1 + nxt.start()] if nxt else text[at:]
with open(EXTENSION, encoding="utf-8") as _f:
_TEXT = _f.read()
REASONS = [line.split("`")[1] for line in section(_TEXT, "## 11. Reasons").splitlines() if line.startswith("| `")]
DOMAINS_STATED = all(('"%s\\x01"' % d[:-1].decode()) in section(_TEXT, "## 2. Records") for d in (SIGN_DOMAIN, LINK_DOMAIN))
# ---------------------------------------------------------------- self-tests
_ED_VERIFY = ED.verify
_ED_CACHE = {}
def _cached_verify(pk, msg, sig, skip=()):
"""ED.verify, remembered: the mutation runs verify the same signatures again and again."""
k = (bytes(pk), bytes(msg), bytes(sig), tuple(skip))
if k not in _ED_CACHE:
_ED_CACHE[k] = _ED_VERIFY(pk, msg, sig, skip)
return _ED_CACHE[k]
def flip(h, at):
b = bytearray(bytes.fromhex(h))
b[at] ^= 0x01
return b.hex()
def rewrite(text, f):
v = json.loads(text)
f(v)
return json.dumps(v)
def mutations(doc):
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
def mutate(label, f):
d = copy.deepcopy(doc)
cs = {c["name"]: c for c in d["cases"]}
f(cs, d)
return label, d
def withhold_second_window(cs, d):
d["admission_logs"]["rated_room"] = [a for a in d["admission_logs"]["rated_room"] if a["t"] < 3600 * 1000000]
def move_admission(cs, d):
d["admission_logs"]["rated_room"][0]["t"] = 3650 * 1000000
def grant_age_byte(cs, d):
g = bytes.fromhex(cs["minors_room"]["grants"][1])
at = g.index(b"\x03age\x02\x0013")
cs["minors_room"]["grants"][1] = (g[:at] + b"\x03age\x02\x0016" + g[at + 8:]).hex()
return [
mutate("a byte of the 16 token's authenticator", lambda cs, d: cs["admit_16"].update(token=flip(cs["admit_16"]["token"], 200))),
mutate("a byte of the 13 token's nonce", lambda cs, d: cs["admit_13"].update(token=flip(cs["admit_13"]["token"], 5))),
mutate("the 13 token's nonce among those redeemed",
lambda cs, d: cs["admit_13"]["redeemed"].append(cs["admit_13"]["token"][4:68])),
mutate("the age asked of the 18 token raised to 19", lambda cs, d: cs["admit_18_where_16"].update(required=19)),
mutate("the 16 token's session challenges withheld", lambda cs, d: cs["admit_16"].update(challenges=[])),
mutate("the policy's issuers given in each other's place",
lambda cs, d: cs["admit_13"].update(policy=rewrite(cs["admit_13"]["policy"], lambda v: v["issuers"].reverse()))),
mutate("the 13 token stated admitted at 16", lambda cs, d: cs["admit_13"]["expect"].update(threshold=16)),
mutate("the programme's claim withheld from the chain", lambda cs, d: cs["derivation_chain"]["claims"].pop(0)),
mutate("a min_age of the transcode's claim lowered",
lambda cs, d: cs["derivation_chain"]["claims"].__setitem__(2, rewrite(cs["derivation_chain"]["claims"][2],
lambda v: v["ratings"][0].update(min_age=12)))),
mutate("a rating dropped from the clip's claim",
lambda cs, d: cs["derivation_chain"]["claims"].__setitem__(4, rewrite(cs["derivation_chain"]["claims"][4],
lambda v: v["ratings"].pop(1)))),
mutate("the clip's age stated as 16", lambda cs, d: cs["derivation_chain"]["expect"]["claims"][4].update(age=16)),
mutate("the transcode stated as not derived", lambda cs, d: cs["derivation_chain"]["expect"]["claims"][2].update(derived=0)),
mutate("the inconsistent claims refused at the first", lambda cs, d: cs["one_value_two_ages"]["expect"].update(at=["claim", 0])),
mutate("the 13-where-16 refusal stated as a replay", lambda cs, d: cs["13_where_16"]["expect"].update(reason="token_replayed")),
mutate("the unrated track's admission stated refused",
lambda cs, d: cs["admit_unrated"].update(expect={"outcome": "refuse", "reason": "below_threshold", "at": ["token"]})),
mutate("the span's claim removed from a request's held", lambda cs, d: cs["rises_for_a_span"]["requests"][3].update(held=[0])),
mutate("a request's threshold raised from 13 to 18", lambda cs, d: cs["rises_for_a_span"]["requests"][3].update(threshold=18)),
mutate("the track's owner stated as the viewer",
lambda cs, d: cs["rises_for_a_span"]["track"]["namespace"].__setitem__(1, d["principals"]["viewer"])),
mutate("the sink's 16 credential withheld", lambda cs, d: cs["holds_13_and_16"]["credentials"].pop(1)),
mutate("the anchor's 16 key stated at 18", lambda cs, d: cs["holds_13_and_16"]["anchor"][1].update(threshold=18)),
mutate("the span's claim withheld from the sink", lambda cs, d: cs["holds_18"]["claims"].pop(1)),
mutate("the second window's admissions withheld from the log", withhold_second_window),
mutate("an admission moved into the second window", move_admission),
mutate("a receipt cell's count changed",
lambda cs, d: cs["without_admissions"].update(receipt=rewrite(cs["without_admissions"]["receipt"],
lambda v: v["cells"][1].update(admitted=60)))),
mutate("the counted cell stated as 49", lambda cs, d: cs["two_windows"]["expect"]["cells"][1].update(admitted=49)),
mutate("the miscounted receipt given without its admissions", lambda cs, d: cs["count_off_by_one"].update(admissions=None)),
mutate("a byte of a viewer's grant attribute age", grant_age_byte),
mutate("the viewer's threshold stated as 16", lambda cs, d: cs["minors_room"]["expect"]["thresholds"].__setitem__(1, 16)),
mutate("the minors' room stated not to meet the profile", lambda cs, d: cs["minors_room"]["expect"].update(meets_profile=0)),
mutate("the minor's grant stated as not having taken effect", lambda cs, d: cs["minors_room"]["expect"].update(minor_grants=[])),
mutate("the minor's join withheld", lambda cs, d: cs["director_records"]["chains"].__setitem__(0, cs["director_records"]["chains"][0][1:])),
mutate("the narrowed grant's refusal stated at the minor's join",
lambda cs, d: cs["grant_narrowed_to_watch_and_chat"]["expect"].update(at=["record", 0, 0])),
mutate("a byte of the signature of the clock's proof of its key",
lambda cs, d: cs["room_service"]["requests"][0]["service"].update(proof=flip(cs["room_service"]["requests"][0]["service"]["proof"], 150))),
mutate("the session with no proof stated as the room's service",
lambda cs, d: cs["room_service"]["expect"]["requests"][3].update(outcome="serve", service=True)),
mutate("the performer's proof stated as a room key's",
lambda cs, d: cs["room_service"]["expect"]["requests"][5].update(outcome="serve", service=True)),
mutate("the forged proof stated as the clock's service",
lambda cs, d: cs["room_service"]["expect"]["requests"][7].update(outcome="serve", service=True)),
mutate("an unattributed claim stated as covering the request",
lambda cs, d: cs["unattributed_claims"]["expect"]["requests"][0].update(age=18, outcome="refuse", reason="below_threshold")),
mutate("the anchor's 16 key listed under its own issuer",
lambda cs, d: cs["key_listed_for_another_issuer"]["anchor"][1].update(issuer="issuer-a.example")),
mutate("the altered credential stated as holding", lambda cs, d: cs["credential_altered"]["expect"].update(state=16)),
mutate("the first window's receipt given the admissions after its span",
lambda cs, d: [a.update(t=3500 * 1000000) for a in d["admission_logs"]["rated_room"] if a["t"] >= 3600 * 1000000]),
mutate("a byte of RFC 9578 A.2's token", lambda cs, d: d["rfc9578_a2_vector1"].update(token=flip(d["rfc9578_a2_vector1"]["token"], 300))),
mutate("a byte of RFC 9578 A.2's blind", lambda cs, d: d["rfc9578_a2_vector1"].update(blind=flip(d["rfc9578_a2_vector1"]["blind"], 40))),
mutate("a byte of RFC 9578 A.2's token response",
lambda cs, d: d["rfc9578_a2_vector1"].update(token_response=flip(d["rfc9578_a2_vector1"]["token_response"], 7))),
mutate("over HTTP, the service stated to be another origin", lambda cs, d: cs["http_admit_16"].update(origin="other.example")),
mutate("over HTTP, a case read as over MOQT", lambda cs, d: cs["http_admit_listed"].update(origin=None)),
mutate("the previous revision's three scopes stated malformed",
lambda cs, d: cs["scope_previous_three"]["expect"].update(reason="scope_malformed")),
mutate("a byte of RFC 9578 A.2's challenge",
lambda cs, d: d["rfc9578_a2_vector1"].update(token_challenge=flip(d["rfc9578_a2_vector1"]["token_challenge"], 10))),
]
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 []
ED.verify = _cached_verify
ok = len(REASONS) == 20 and len(set(REASONS)) == len(REASONS) and DOMAINS_STATED
print(("PASS" if ok else "FAIL") + " §11 read from the extension: %d reasons; §2's domains as stated" % len(REASONS))
fails += [] if ok else ["§11 or §2 does not read as this verifier expects"]
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
doc = json.load(f)
a2 = a2_vector(doc["rfc9578_a2_vector1"])
print(("PASS" if not a2 else "FAIL") + " RFC 9578 Appendix A.2 vector 1: challenge digest, key id and RSASSA-PSS%s"
% ("" if not a2 else " (%s)" % "; ".join(a2)))
fails += ["RFC 9578 A.2 vector 1: " + f for f in a2]
keys = issuer_key_facts(doc)
print(("PASS" if not keys else "FAIL") + " the issuer keys read by §4.2, and the 18 key is RFC 9578 A.2's")
fails += keys
seen = set()
for c in doc["cases"]:
got = run(c, doc)
ok = got == c["expect"]
if c["expect"].get("reason"):
seen.add(c["expect"]["reason"])
if c["kind"] == "serving" and ok:
seen |= {q.get("reason") for q in c["expect"].get("requests", [])}
what = got["outcome"] + (" %s at %s" % (got["reason"], got["at"]) if "reason" in got else "")
print(("PASS" if ok else "FAIL") + " %s %s: %s" % (c["kind"], c["name"], what))
if not ok:
fails.append("%s: %s, the corpus states %s" % (c["name"], json.dumps(got)[:400], json.dumps(c["expect"])[:400]))
missing = [r for r in REASONS if r not in seen]
print(("PASS" if not missing else "FAIL") + " every reason of §11 refuses a case" + (": not %s" % ", ".join(missing) if missing else ""))
fails += ["no case refused with %s" % r for r in missing]
# A copy is caught only by a failure the corpus itself does not already give.
base = set(check(doc))
muts = mutations(doc)
for label, d in muts:
caught = [f for f in check(d) if f not in base]
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:110] if caught else ""))
if not caught:
fails.append("a mutated copy verifies: %s" % label)
gaps = []
for name, (expected, what) in CHECKS.items():
DROPPED.clear()
DROPPED.add(name)
try:
caught = [f for f in check(doc) if f not in base]
finally:
DROPPED.clear()
if expected:
print(("PASS" if caught else "FAIL") + " verifier without check %s (%s) is caught%s"
% (name, what, " (%s)" % caught[0][:90] if caught else ""))
if not caught:
fails.append("the corpus does not catch a verifier without %s" % name)
gaps.append(name)
elif caught:
print("PASS verifier without check %s (%s) is caught (%s)" % (name, what, caught[0][:90]))
else:
gaps.append(name)
for name in gaps:
print("NOTE no case reaches check %s (%s)" % (name, CHECKS[name][1]))
for f in fails:
print("FAIL " + f)
print(("PASS" if not fails else "FAIL") + " age-assurance corpus: %d cases, %d mutated copies, %d of %d dropped checks caught"
% (len(doc["cases"]), len(muts), len(CHECKS) - len(gaps), len(CHECKS)))
return 1 if fails else 0
if __name__ == "__main__":
if len(sys.argv) != 2:
print("usage: age_assurance_verify.py <live-conformance/age>", file=sys.stderr)
sys.exit(2)
sys.exit(main(sys.argv[1]))
Reference in a new issue View git blame Copy permalink