1a8c1a5ead
live-rooms §10.4's subscribing scopes did not fit the 255 bytes draft-ietf-moq-privacy-pass-auth-03 gives one MoQAuthorizationInfo: the room's scope is now one MoQAuthScope, the subscribing actions on MATCH_PREFIX of [R] and on every track name, 78 bytes. A challenge's origin_info is read by its binding: over MOQT the room's scope byte for byte (scope_malformed, scope_mismatch, the previous revision's scopes among them); over HTTP RFC 9577's list of server names holding the service's own. The corpus gains the scope and HTTP cases and RFC 9578 A.2's issuance, which both verifiers check. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1554 lines
69 KiB
Python
1554 lines
69 KiB
Python
#!/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), 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"),
|
||
# §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)
|
||
if 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}
|
||
|
||
|
||
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("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]))
|