Security · Check it yourself
verify-audit5.py: check an audit export without MoxiVue
A script in plain Python that checks an exported account log from its first entry: every signature, the chain, both witnesses' cosignatures and every rule about who may do what. It contains no MoxiVue code and needs nothing installed besides Python 3, so you do not have to take our program's word for it.
Download verify-audit5.py · 4228 lines · SHA-256 31788531c0408379957b5d8d66ef4d31d335acd7900164dfcca3fa9bb7a4e8c2
Run it
moxivue audit export --dir audit-2026
python3 verify-audit5.py audit-2026/log.mxlog --records audit-2026/records.canonIt prints how many entries it verified from the first one and how many are in checkpoints both witnesses cosigned, and with --records how many machine records it checked against the log and which are missing; exit status 0 means verified, 1 not verified (it names the first failing entry and its rule), 2 a usage error. --json prints the result as JSON. What is in an export, and what a check shows and does not, are on the audit export page. The witnesses' public keys are written into the script with each release, as they are built into the apps; without them it refuses every export rather than accept one.
Read it before you run it: that is the point. The listing below is the file itself, byte for byte; after downloading, sha256sum verify-audit5.py must print the value above. That value comes from this website too: it shows the file was not changed on the way, not that our server is honest.
The script
#!/usr/bin/env python3
"""Check an exported MoxiVue 5.0 account log without MoxiVue.
python3 verify-audit5.py <export> [--json]
Uses only Python's standard library and no MoxiVue code (architecture decision C and decision 16): anyone can read
and run it. It applies the same rules as MoxiVue's own rules core (crates/moxivue-log), and is checked against it on
shared test logs and vectors (the push checks run both).
What it checks, from the first entry (the genesis) to the last:
- the encoding: every structure is the one canonical encoding (a strict subset of deterministic CBOR), with
exactly the keys its type has; unknown kinds, suites, format versions and algorithm tags are refused;
- the chain: every leaf names its position and the hash of the leaf before it; the Merkle tree (RFC 6962) of the
leaves is the one the witnesses' checkpoints cover, and a checkpoint whose root differs is a fork;
- the signatures: every signature is hybrid (Ed25519 and ML-DSA-65), and counts only when both halves verify
over the same labelled message. Ed25519 is verified strictly (RFC 8032 with canonical encodings, no small-order
R, s below the group order, keys of neither small nor mixed order), ML-DSA-65 as FIPS 204 Algorithm 8 (hint
indices strictly increasing: repeated or unordered hints are refused). Both are written out below;
- the keys: X-Wing keys half by half (the ML-KEM-768 modulus check; the X25519 half canonical, not of low order,
not on the twist), every key's proof of possession, no public key half twice in the account's log;
- every entry's base checkpoint, cosigned by both witnesses with stamps within 5 minutes of each other;
- every log rule: who may propose, confirm, refuse, withdraw and finish which change (roles, reach, persons,
freezes, the set-up shape and the floor, conflicts after a base, lapses), generations (chain, reason, exact
recipient lists), joins (transcript fields, inviting key), machine and token joins (one use, expiry, revocation),
settings (versions, sensitive changes), SSH grants, recovery (snapshot, named devices, cancels, both witnesses'
cancel lists, the delay), places, rotations and support statements, replays (entry ids).
- with --records, the machines' records and status exported with the log (moxivue-data's records export): each
names this account and a generation of the log, is signed by the record key (K4) its machine held while that
generation was current, comes from a machine that was in the record's place then, and follows the sequences
before it (a replay or reordering is refused); skipped record sequences are listed as missing. It does not open
them (it holds no keys).
An authentic entry that breaks a rule stays in the log (a cosigned log cannot drop it) with no effect; the check
names it with its rule and the export is not verified.
Exit codes: 0 verified (no failure and no refused entry); 1 not verified (the first failing or refused entry and its
rule are printed); 2 usage or an unreadable file. Running it again changes nothing (it only reads).
Other uses (for the push checks and for anyone who wants to check this script):
python3 verify-audit5.py --test-log <golden-file> [--json] the command on a test log, with its test witnesses
python3 verify-audit5.py --self-test <vectors-dir> the published Ed25519, X25519, ML-KEM and ML-DSA vectors
python3 verify-audit5.py --golden <file-or-dir>... the test logs of the rules core: same verdict, same rule
python3 verify-audit5.py --shared <shared.canon> the shared signed vectors (the same file Rust checks)
python3 verify-audit5.py --records-vectors <records.canon> records checked as moxivue-data's console does
python3 verify-audit5.py --mutate <golden-file> [n] every sampled one-byte change of a test log is refused
python3 verify-audit5.py --speed <golden-file> time to verify a test log
The witnesses: checkpoints are C2SP checkpoints with both witnesses' hybrid cosignatures (C2SP cosignature/v1). Their
keys are filled in with the rollout that deploys them (WITNESS_KEYS below, the same as the apps' witnesses.txt); until
then no checkpoint verifies and every export past its genesis is refused (never accepted). Test logs carry the test
witnesses' keys and are read only with --test-log, --golden, --mutate and --speed.
"""
import base64
import bisect
import hashlib
import json
import os
import sys
import time
from array import array
from collections import namedtuple
# ======================================================================== canonical encoding
# The same encoding as crates/moxivue-proto/src/canon.rs (and tools/moxivue_canon.py): integers in the shortest
# head, byte and text strings, arrays, maps with text keys in canonical order, false/true/null; nothing else.
MAX_DEPTH = 32
INT_MIN = -(1 << 64)
INT_MAX = (1 << 64) - 1
class CanonError(Exception):
pass
def _head(major, n):
m = major << 5
if n < 24:
return bytes([m | n])
if n <= 0xFF:
return bytes([m | 24, n])
if n <= 0xFFFF:
return bytes([m | 25]) + n.to_bytes(2, "big")
if n <= 0xFFFFFFFF:
return bytes([m | 26]) + n.to_bytes(4, "big")
return bytes([m | 27]) + n.to_bytes(8, "big")
def _key_order(k):
b = k.encode("utf-8")
return (len(b), b)
def _enc(v, depth, out):
if v is False:
out.append(b"\xf4")
elif v is True:
out.append(b"\xf5")
elif v is None:
out.append(b"\xf6")
elif type(v) is int:
if not INT_MIN <= v <= INT_MAX:
raise CanonError("IntRange")
out.append(_head(0, v) if v >= 0 else _head(1, -1 - v))
elif type(v) is bytes:
out.append(_head(2, len(v)))
out.append(v)
elif type(v) is str:
b = v.encode("utf-8")
out.append(_head(3, len(b)))
out.append(b)
elif type(v) is list:
if depth >= MAX_DEPTH:
raise CanonError("EncodeTooDeep")
out.append(_head(4, len(v)))
for x in v:
_enc(x, depth + 1, out)
elif type(v) is dict:
if depth >= MAX_DEPTH:
raise CanonError("EncodeTooDeep")
out.append(_head(5, len(v)))
for k in sorted(v, key=_key_order):
if type(k) is not str:
raise CanonError("KeyNotText")
_enc(k, depth + 1, out)
_enc(v[k], depth + 1, out)
else:
raise CanonError("Type")
def cenc(v):
out = []
_enc(v, 0, out)
return b"".join(out)
class _Dec:
__slots__ = ("b", "at")
def __init__(self, b):
self.b = b
self.at = 0
def take(self, n):
if n > len(self.b) - self.at:
raise CanonError("Truncated")
s = self.b[self.at:self.at + n]
self.at += n
return s
def head(self):
ib = self.take(1)[0]
major, ai = ib >> 5, ib & 0x1F
if major == 7:
return 7, ai
if ai < 24:
return major, ai
if ai in (28, 29, 30):
raise CanonError("Reserved")
if ai == 31:
raise CanonError("Indefinite")
size = {24: 1, 25: 2, 26: 4, 27: 8}[ai]
n = int.from_bytes(self.take(size), "big")
if n < {24: 24, 25: 0x100, 26: 0x10000, 27: 0x100000000}[ai]:
raise CanonError("NotShortest")
return major, n
def text(self, n):
try:
return self.take(n).decode("utf-8")
except UnicodeDecodeError:
raise CanonError("Utf8")
def value(self, depth):
major, n = self.head()
if major == 0:
return n
if major == 1:
return -1 - n
if major == 2:
return bytes(self.take(n))
if major == 3:
return self.text(n)
if major in (4, 5):
if depth >= MAX_DEPTH:
raise CanonError("TooDeep")
if n > len(self.b) - self.at:
raise CanonError("Truncated")
if major == 4:
return [self.value(depth + 1) for _ in range(n)]
m = {}
prev = None
for _ in range(n):
km, kn = self.head()
if km != 3:
raise CanonError("KeyNotText")
k = self.text(kn)
if prev is not None:
if _key_order(prev) == _key_order(k):
raise CanonError("KeyRepeated")
if _key_order(prev) > _key_order(k):
raise CanonError("KeyOrder")
m[k] = self.value(depth + 1)
prev = k
return m
if major == 6:
raise CanonError("Tag")
if n == 20:
return False
if n == 21:
return True
if n == 22:
return None
if n == 31:
raise CanonError("Indefinite")
raise CanonError("Simple")
def cdec(b):
d = _Dec(bytes(b))
v = d.value(0)
if d.at != len(d.b):
raise CanonError("Trailing")
return v
def labelled(label, body):
"""encode([label, bytes(body)]): what every signature and labelled hash is over (key chapter R2.5)."""
return cenc([label, body])
def sha256(b):
return hashlib.sha256(b).digest()
# ======================================================================== Ed25519 (RFC 8032), strict
# As MoxiVue's Rust side (ed25519-dalek 2.2 verify_strict under a key checked by moxivue-proto check.rs):
# - a key is the canonical encoding of a point (y < p; x = 0 never with the sign bit) of neither small nor mixed
# order ([8]A is not the neutral point, [L]A is);
# - a signature: s < L; R decodes to a point that is not of small order; and the encoding of [s]B - [k]A equals R's
# bytes (the cofactorless equation; a non-canonical R can never equal it), k = SHA-512(R || A || M) mod L.
EP = 2 ** 255 - 19
EL = 2 ** 252 + 27742317777372353535851937790883648493
ED = -121665 * pow(121666, EP - 2, EP) % EP
ED2 = 2 * ED % EP
SQRT_M1 = pow(2, (EP - 1) // 4, EP)
_ZERO = (0, 1, 1, 0)
def _ed_add(a, b):
x1, y1, z1, t1 = a
x2, y2, z2, t2 = b
A = (y1 - x1) * (y2 - x2) % EP
B = (y1 + x1) * (y2 + x2) % EP
C = t1 * ED2 % EP * t2 % EP
Dd = 2 * z1 * z2 % EP
E, F, G, H = B - A, Dd - C, Dd + C, B + A
return (E * F % EP, G * H % EP, F * G % EP, E * H % EP)
def _ed_double(a):
x1, y1, z1, _ = a
A = x1 * x1 % EP
B = y1 * y1 % EP
C = 2 * z1 * z1 % EP
H = A + B
xy = x1 + y1
E = H - xy * xy % EP
G = A - B
F = C + G
return (E * F % EP, G * H % EP, F * G % EP, E * H % EP)
def _ed_neg(a):
x, y, z, t = a
return ((-x) % EP, y, z, (-t) % EP)
def _ed_is_zero(a):
return a[0] % EP == 0 and (a[1] - a[2]) % EP == 0
def _ed_mul(s, pt):
q = _ZERO
while s > 0:
if s & 1:
q = _ed_add(q, pt)
pt = _ed_double(pt)
s >>= 1
return q
def _ed_encode(a):
x, y, z, _ = a
zi = pow(z, EP - 2, EP)
x, y = x * zi % EP, y * zi % EP
return (y | ((x & 1) << 255)).to_bytes(32, "little")
def _ed_decode(s):
"""The point of a canonical 32-byte encoding, or None."""
if len(s) != 32:
return None
y = int.from_bytes(s, "little")
sign = y >> 255
y &= (1 << 255) - 1
if y >= EP:
return None
u = (y * y - 1) % EP
v = (ED * y * y + 1) % EP
x2 = u * pow(v, EP - 2, EP) % EP
if x2 == 0:
if sign:
return None
return (0, y, 1, 0)
x = pow(x2, (EP + 3) // 8, EP)
if (x * x - x2) % EP != 0:
x = x * SQRT_M1 % EP
if (x * x - x2) % EP != 0:
return None
if (x & 1) != sign:
x = EP - x
return (x, y, 1, x * y % EP)
_EB = _ed_decode(bytes.fromhex("5866666666666666666666666666666666666666666666666666666666666666"))
# [j * 16^i] B for i < 64, j < 16: [s]B with 64 additions
_EB_TABLE = []
def _eb_table():
if not _EB_TABLE:
p = _EB
for _ in range(64):
row = [_ZERO]
for _ in range(15):
row.append(_ed_add(row[-1], p))
_EB_TABLE.append(row)
for _ in range(4):
p = _ed_double(p)
return _EB_TABLE
def _ed_small_order(p):
return _ed_is_zero(_ed_double(_ed_double(_ed_double(p))))
class EdKey:
"""An accepted Ed25519 key: canonical, neither small nor mixed order."""
__slots__ = ("raw", "neg_table")
def __init__(self, raw, point):
self.raw = raw
neg = _ed_neg(point)
t = [_ZERO]
for _ in range(15):
t.append(_ed_add(t[-1], neg))
self.neg_table = t
def accept_ed25519(pk):
"""EdKey, or the reason it is refused (R2.4)."""
if len(pk) != 32:
return "Length"
p = _ed_decode(pk)
if p is None:
return "EdNotAPointOrNonCanonical"
if _ed_small_order(p):
return "EdSmallOrder"
if not _ed_is_zero(_ed_mul(EL, p)):
return "EdMixedOrder"
return EdKey(bytes(pk), p)
def ed25519_verify_strict(key, msg, sig):
if len(sig) != 64:
return False
rb = sig[:32]
s = int.from_bytes(sig[32:], "little")
if s >= EL:
return False
r = _ed_decode(rb)
if r is None or _ed_small_order(r):
return False
k = int.from_bytes(hashlib.sha512(rb + key.raw + msg).digest(), "little") % EL
tb = _eb_table()
acc = _ZERO
for i in range(64):
n = (s >> (4 * i)) & 15
if n:
acc = _ed_add(acc, tb[i][n])
# [k](-A), four bits at a time from the top
q = _ZERO
nt = key.neg_table
for i in range(63, -1, -1):
q = _ed_double(_ed_double(_ed_double(_ed_double(q))))
n = (k >> (4 * i)) & 15
if n:
q = _ed_add(q, nt[n])
return _ed_encode(_ed_add(acc, q)) == rb
# ======================================================================== ML-DSA-65 (FIPS 204), verification only
MQ = 8380417
MD = 13
TAU = 49
CT_LEN = 48 # lambda / 4
GAMMA1 = 1 << 19
GAMMA2 = (MQ - 1) // 32
MK, ML_ = 6, 5
BETA = TAU * 4
OMEGA = 55
MLDSA_VK_LEN = 32 + 32 * MK * 10
MLDSA_SIG_LEN = CT_LEN + ML_ * 32 * 20 + OMEGA + MK
def _brv8(k):
return int("{:08b}".format(k)[::-1], 2)
ZETAS = [pow(1753, _brv8(k), MQ) for k in range(256)]
def ntt(f):
"""FIPS 204 Algorithm 41."""
w = list(f)
m = 0
ln = 128
while ln >= 1:
start = 0
while start < 256:
m += 1
z = ZETAS[m]
for j in range(start, start + ln):
t = z * w[j + ln] % MQ
w[j + ln] = (w[j] - t) % MQ
w[j] = (w[j] + t) % MQ
start += 2 * ln
ln //= 2
return w
def intt(w):
"""FIPS 204 Algorithm 42."""
w = list(w)
m = 256
ln = 1
while ln < 256:
start = 0
while start < 256:
m -= 1
z = MQ - ZETAS[m]
for j in range(start, start + ln):
t = w[j]
w[j] = (t + w[j + ln]) % MQ
w[j + ln] = z * (t - w[j + ln]) % MQ
start += 2 * ln
ln *= 2
return [8347681 * x % MQ for x in w]
# Multiplication in Z_q[X]/(X^256 + 1) through one big-integer product (Kronecker substitution, 64-bit slots):
# the same values as NTT^-1(NTT(a) o NTT(b)) (checked by --self-test against the transform above), computed by
# CPython's integer multiplication instead of 1,024 Python-level butterflies.
_NATIVE_LE = sys.byteorder == "little"
def _pack(coeffs):
a = array("Q", coeffs)
if not _NATIVE_LE:
a.byteswap()
return int.from_bytes(a.tobytes(), "little")
def _unpack_fold(x):
"""The negacyclic fold of a packed product (or sum of products) of polynomials with coefficients in [0, q)."""
a = array("Q")
a.frombytes(x.to_bytes(8 * 512, "little"))
if not _NATIVE_LE:
a.byteswap()
lo = a[:256]
hi = a[256:]
return [(x - y) % MQ for x, y in zip(lo, hi)]
def _bits_unpack(b, bits, n=256):
x = int.from_bytes(b, "little")
mask = (1 << bits) - 1
return [(x >> (bits * i)) & mask for i in range(n)]
def _rej_ntt_poly(seed):
"""FIPS 204 Algorithm 30 (RejNTTPoly) with SHAKE128."""
want = 840
while True:
buf = hashlib.shake_128(seed).digest(want)
out = []
for i in range(0, want - 2, 3):
z = buf[i] | (buf[i + 1] << 8) | ((buf[i + 2] & 0x7F) << 16)
if z < MQ:
out.append(z)
if len(out) == 256:
return out
want += 168 * 5
def _sample_in_ball(rho):
"""FIPS 204 Algorithm 29."""
want = 8 + 136 * 2
while True:
buf = hashlib.shake_256(rho).digest(want)
h = int.from_bytes(buf[:8], "little")
c = [0] * 256
pos = 8
ok = True
for i in range(256 - TAU, 256):
while True:
if pos >= want:
ok = False
break
j = buf[pos]
pos += 1
if j <= i:
break
if not ok:
break
c[i] = c[j]
c[j] = -1 if (h >> (i + TAU - 256)) & 1 else 1
if ok:
return c
want += 136 * 4
class MlDsaKey:
"""A decoded ML-DSA-65 public key with what verification reuses: tr, A (as polynomials) and -t1 * 2^d."""
__slots__ = ("pk", "tr", "a_packed", "t1_packed")
def __init__(self, pk):
assert len(pk) == MLDSA_VK_LEN
self.pk = bytes(pk)
a_hat, t1 = self.expand()
self.a_packed = [[_pack(intt(a_hat[r][s])) for s in range(ML_)] for r in range(MK)]
self.t1_packed = [_pack([(-(x << MD)) % MQ for x in t1[r]]) for r in range(MK)]
self.tr = hashlib.shake_256(pk).digest(64)
def expand(self):
"""(A in the NTT domain by FIPS 204 Algorithm 32 ExpandA, t1 by Algorithm 23 pkDecode)"""
rho = self.pk[:32]
t1 = [_bits_unpack(self.pk[32 + 320 * i:32 + 320 * (i + 1)], 10) for i in range(MK)]
return [[_rej_ntt_poly(rho + bytes([s, r])) for s in range(ML_)] for r in range(MK)], t1
def mldsa_sig_decode(sig):
"""(c_tilde, z, h) of FIPS 204 Algorithm 27, or None (Algorithm 21's hint checks: indices strictly increasing
within each polynomial, counts in order and at most omega, unused positions zero)."""
if len(sig) != MLDSA_SIG_LEN:
return None
ct = sig[:CT_LEN]
z = []
at = CT_LEN
for _ in range(ML_):
z.append([GAMMA1 - x for x in _bits_unpack(sig[at:at + 640], 20)])
at += 640
y = sig[at:]
h = [[0] * 256 for _ in range(MK)]
index = 0
for i in range(MK):
if y[OMEGA + i] < index or y[OMEGA + i] > OMEGA:
return None
first = index
while index < y[OMEGA + i]:
if index > first and y[index - 1] >= y[index]:
return None
h[i][y[index]] = 1
index += 1
for i in range(index, OMEGA):
if y[i] != 0:
return None
return ct, z, h
def _use_hint_row(hrow, r):
"""FIPS 204 Algorithms 36 (Decompose) and 40 (UseHint) over one polynomial, values r in [0, q)."""
out = []
g2 = 2 * GAMMA2
for hi, ri in zip(hrow, r):
r0 = ri % g2
if r0 > GAMMA2:
r0 -= g2
if ri - r0 == MQ - 1:
r1 = 0
r0 -= 1
else:
r1 = (ri - r0) // g2
if hi:
r1 = (r1 + 1) % 16 if r0 > 0 else (r1 - 1) % 16
out.append(r1)
return out
def mldsa_verify_mu(key, mu, sig, reference=False):
"""FIPS 204 Algorithm 8 from mu on. reference=True takes the transform path (for the self-test)."""
d = mldsa_sig_decode(sig)
if d is None:
return False
return _mldsa_check(key, mu, d, reference)
def _mldsa_check(key, mu, d, reference=False):
ct, z, h = d
bound = GAMMA1 - BETA
for zi in z:
for x in zi:
if x >= bound or x <= -bound:
return False
c = _sample_in_ball(ct)
if reference:
a_hat, t1 = key.expand()
zh = [ntt([x % MQ for x in zi]) for zi in z]
ch = ntt([x % MQ for x in c])
w = []
for r in range(MK):
acc = [0] * 256
for s in range(ML_):
a = a_hat[r][s]
zs = zh[s]
acc = [(x + y * u) % MQ for x, y, u in zip(acc, a, zs)]
t = ntt([(x << MD) % MQ for x in t1[r]])
acc = [(x - y * u) % MQ for x, y, u in zip(acc, ch, t)]
w.append(intt(acc))
else:
zp = [_pack([x % MQ for x in zi]) for zi in z]
cp = _pack([x % MQ for x in c])
w = []
for r in range(MK):
row = key.a_packed[r]
acc = cp * key.t1_packed[r]
for s in range(ML_):
acc += row[s] * zp[s]
w.append(_unpack_fold(acc))
w1 = bytearray()
for r in range(MK):
u = _use_hint_row(h[r], w[r])
w1 += bytes(u[2 * i] | (u[2 * i + 1] << 4) for i in range(128))
return hashlib.shake_256(mu + bytes(w1)).digest(CT_LEN) == ct
def mldsa_verify(key, msg, sig, ctx=b"", internal=False):
"""ML-DSA.Verify (FIPS 204 Algorithm 3) with context ctx; internal=True is Algorithm 8 on M' = msg."""
if internal:
mp = msg
else:
if len(ctx) > 255:
return False
mp = bytes([0, len(ctx)]) + ctx + msg
mu = hashlib.shake_256(key.tr + mp).digest(64)
return mldsa_verify_mu(key, mu, sig)
# ======================================================================== X25519, ML-KEM-768 and X-Wing key checks
# As moxivue-proto check.rs: the X25519 half canonical (bit 255 clear, u < p), not of low order (a clamped scalar
# times it is not zero), not on the twist (it maps to an Edwards point); the ML-KEM-768 encapsulation key passes the
# FIPS 203 modulus check (every 12-bit coefficient below 3329).
A24 = 121665
def _x25519(k, u):
"""RFC 7748 X25519 with the scalar k (bytes, clamped here) on the canonical u-coordinate u (int)."""
k = bytearray(k)
k[0] &= 248
k[31] &= 127
k[31] |= 64
k = int.from_bytes(k, "little")
x1 = u
x2, z2, x3, z3 = 1, 0, u, 1
swap = 0
for t in range(254, -1, -1):
kt = (k >> t) & 1
swap ^= kt
if swap:
x2, x3, z2, z3 = x3, x2, z3, z2
swap = kt
A = (x2 + z2) % EP
AA = A * A % EP
B = (x2 - z2) % EP
BB = B * B % EP
E = (AA - BB) % EP
C = (x3 + z3) % EP
D_ = (x3 - z3) % EP
DA = D_ * A % EP
CB = C * B % EP
x3 = (DA + CB) ** 2 % EP
z3 = x1 * (DA - CB) ** 2 % EP
x2 = AA * BB % EP
z2 = E * (AA + A24 * E) % EP
if swap:
x2, x3, z2, z3 = x3, x2, z3, z2
return x2 * pow(z2, EP - 2, EP) % EP
def accept_x25519(u):
"""None when accepted, else the reason."""
if len(u) != 32:
return "Length"
n = int.from_bytes(u, "little")
if u[31] & 0x80 or n >= EP:
return "X25519NonCanonical"
if _x25519(bytes([0x5A]) * 32, n) == 0:
return "X25519LowOrder"
# the Edwards point y = (u - 1) / (u + 1) must exist (sign 0): else u is on the twist
if (n + 1) % EP == 0:
return "X25519Twist"
y = (n - 1) * pow(n + 1, EP - 2, EP) % EP
x2 = (y * y - 1) * pow(ED * y * y + 1, EP - 2, EP) % EP
if x2 != 0 and pow(x2, (EP - 1) // 2, EP) != 1:
return "X25519Twist"
return None
MLKEM768_EK_LEN = 1184
XWING_PK_LEN = MLKEM768_EK_LEN + 32
XWING_CT_LEN = 1088 + 32
def accept_mlkem768(ek):
if len(ek) != MLKEM768_EK_LEN:
return "Length"
b = ek
for i in range(0, 1152, 3):
if (b[i] | ((b[i + 1] & 0x0F) << 8)) >= 3329 or ((b[i + 1] >> 4) | (b[i + 2] << 4)) >= 3329:
return "MlKemModulus"
return None
def accept_xwing_pk(pk):
if len(pk) != XWING_PK_LEN:
return "Length"
return accept_mlkem768(pk[:MLKEM768_EK_LEN]) or accept_x25519(pk[MLKEM768_EK_LEN:])
# ======================================================================== algorithm tags, hybrid keys and signatures
# Every key, signature and sealed item starts with a two-byte algorithm tag (big-endian); the 5.0 suite (1) knows
# exactly these and no other.
TAG_SIG = b"\x01\x01" # Ed25519 + ML-DSA-65, both must verify
TAG_KEM = b"\x02\x01" # X-Wing (ML-KEM-768 + X25519)
TAG_SEAL = b"\x03\x01" # HPKE (X-Wing, HKDF-SHA256, ChaCha20-Poly1305)
KNOWN_TAGS = (TAG_SIG, TAG_KEM, TAG_SEAL, b"\x04\x01", b"\x05\x01")
ED_SIG_LEN = 64
SIG_KEY_LEN = 2 + 32 + MLDSA_VK_LEN
SIG_LEN = 2 + ED_SIG_LEN + MLDSA_SIG_LEN
SUITE = 1
FORMAT = 1
# the labels of key chapter R2.5 that sign (moxivue-proto label.rs, use Sign); a signature under any other is refused
SIGN_LABELS = frozenset("moxivue/%s/v1" % p for p in (
"entry", "proposal", "confirmation", "rotation", "support", "cancel", "recovery-request", "root-request",
"root-result", "record", "status", "channel-bind", "cancel-list", "witness-answer", "cancel-receipt",
"token-join", "executor-result", "executor-audit", "recording", "dispatcher-login", "witness-submit",
"recording-reseal", "action-record"))
L_ENTRY, L_PROPOSAL, L_CONFIRMATION = "moxivue/entry/v1", "moxivue/proposal/v1", "moxivue/confirmation/v1"
L_ROTATION, L_SUPPORT, L_CANCEL = "moxivue/rotation/v1", "moxivue/support/v1", "moxivue/cancel/v1"
L_RECOVERY_REQUEST, L_TOKEN_JOIN = "moxivue/recovery-request/v1", "moxivue/token-join/v1"
L_WITNESS_ANSWER, L_CANCEL_LIST = "moxivue/witness-answer/v1", "moxivue/cancel-list/v1"
L_PAIRING_TRANSCRIPT, L_WORDS = "moxivue/pairing-transcript/v1", "moxivue/words/v1"
def _tag(b, want):
"""None when b starts with the tag `want`, else the reason."""
if len(b) < 2:
return "no algorithm tag"
t = bytes(b[:2])
if t not in KNOWN_TAGS:
return "unknown algorithm tag 0x%s" % t.hex()
if t != want:
return "algorithm tag 0x%s where 0x%s was expected" % (t.hex(), want.hex())
return None
class SigKey:
__slots__ = ("ed", "ml_raw", "_ml")
def __init__(self, ed, ml_raw):
self.ed = ed
self.ml_raw = ml_raw
self._ml = None
@property
def ml(self):
if self._ml is None:
self._ml = MlDsaKey(self.ml_raw)
return self._ml
_SIG_KEYS = {}
def sig_key(b):
"""A hybrid signature key (tag, Ed25519 32, ML-DSA-65 1952), checked (R2.4); SigKey or the reason refused."""
k = _SIG_KEYS.get(b)
if k is not None:
return k
why = _tag(b, TAG_SIG)
if why:
return why
if len(b) != SIG_KEY_LEN:
return "wrong length"
ed = accept_ed25519(b[2:34])
if isinstance(ed, str):
return "weak key: " + ed
k = SigKey(ed, bytes(b[34:]))
_SIG_KEYS[bytes(b)] = k
return k
_KEM_OK = {}
def kem_key(b):
"""None when an X-Wing key (tag, ML-KEM-768 1184, X25519 32) is accepted half by half, else the reason."""
if b in _KEM_OK:
return _KEM_OK[b]
why = _tag(b, TAG_KEM) or accept_xwing_pk(b[2:])
_KEM_OK[bytes(b)] = why
return why
def sealed_ok(b):
"""None when b is a sealed box of the suite's HPKE (tag, X-Wing encapsulation, at least a 16-byte tag)."""
why = _tag(b, TAG_SEAL)
if why:
return why
if len(b) - 2 < XWING_CT_LEN + 16:
return "wrong length"
return None
def hybrid_verify(key_bytes, label, msg, sig_bytes):
"""A hybrid signature over labelled(label, msg): counts only when both halves verify (R2.4)."""
if label not in SIGN_LABELS:
return False
k = sig_key(key_bytes)
if isinstance(k, str):
return False
if _tag(sig_bytes, TAG_SIG) or len(sig_bytes) != SIG_LEN:
return False
ml_sig = sig_bytes[2 + ED_SIG_LEN:]
if mldsa_sig_decode(ml_sig) is None:
return False
m = labelled(label, bytes(msg))
if not ed25519_verify_strict(k.ed, m, sig_bytes[2:2 + ED_SIG_LEN]):
return False
return mldsa_verify(k.ml, m, ml_sig)
def labelled_hash(label, body):
return sha256(labelled(label, body))
# ======================================================================== log structures (crates/moxivue-log)
# Every structure is a map with an exact set of keys and its type in "t"; a missing key, an unknown key or a value
# of the wrong type is refused. A decoding error is a verification failure.
class DecodeError(Exception):
pass
def _bad(why):
raise DecodeError(why)
def _is_u64(v):
return type(v) is int and 0 <= v <= INT_MAX
class Rd:
__slots__ = ("m", "taken", "ctx")
def __init__(self, v, ctx):
if type(v) is not dict:
_bad(ctx + ": not a map")
self.m = v
self.taken = 0
self.ctx = ctx
@staticmethod
def typed(v, t):
r = Rd(v, t)
got = r.text("t")
if got != t:
_bad("type %r where %r was expected" % (got, t))
return r
def get(self, k):
if k not in self.m:
_bad("%s: no %r" % (self.ctx, k))
self.taken += 1
return self.m[k]
def opt(self, k):
if k in self.m:
self.taken += 1
return self.m[k]
return None
def done(self):
if self.taken != len(self.m):
_bad("%s: unknown keys among %r" % (self.ctx, sorted(self.m)))
def u64(self, k):
v = self.get(k)
if not _is_u64(v):
_bad("%s: %r not an unsigned integer" % (self.ctx, k))
return v
def bool(self, k):
v = self.get(k)
if type(v) is not bool:
_bad("%s: %r not a boolean" % (self.ctx, k))
return v
def text(self, k):
v = self.get(k)
if type(v) is not str:
_bad("%s: %r not text" % (self.ctx, k))
return v
def bytes(self, k):
v = self.get(k)
if type(v) is not bytes:
_bad("%s: %r not bytes" % (self.ctx, k))
return v
def hash(self, k):
v = self.get(k)
if type(v) is not bytes or len(v) != 32:
_bad("%s: %r not a 32-byte hash" % (self.ctx, k))
return v
def array(self, k):
v = self.get(k)
if type(v) is not list:
_bad("%s: %r not an array" % (self.ctx, k))
return v
def role(self, k):
w = self.text(k)
if w not in ROLES:
_bad("unknown role %r" % w)
return w
def base(self, k):
return base_from(self.get(k))
def devices(self, k):
out = []
for v in self.array(k):
if not _is_u64(v):
_bad("%r: not a device id" % k)
if out and out[-1] >= v:
_bad("%r: device ids not strictly increasing" % k)
out.append(v)
return out
def _hash_of(v):
return v if type(v) is bytes and len(v) == 32 else None
# a base checkpoint: (size, root, t1, t2)
Base = namedtuple("Base", "size root t1 t2")
def base_later(b):
return max(b.t1, b.t2)
def base_value(b):
return {"t": "base", "size": b.size, "root": b.root, "t1": b.t1, "t2": b.t2}
def base_from(v):
r = Rd.typed(v, "base")
b = Base(r.u64("size"), r.hash("root"), r.u64("t1"), r.u64("t2"))
r.done()
return b
# roles (§3.1): owner > admin > operator > viewer; the auditor beside the viewer, outside that order (AUD-1)
OWNER, ADMIN, OPERATOR, VIEWER, AUDITOR = "owner", "admin", "operator", "viewer", "auditor"
ROLES = (OWNER, ADMIN, OPERATOR, VIEWER, AUDITOR)
_RANK = {OWNER: 3, ADMIN: 2, OPERATOR: 1, VIEWER: 0, AUDITOR: None}
def rank(role):
return _RANK[role]
def _rank_cmp(role):
"""the rank as a number where no rank (the auditor) is below every rank (Rust's Option order)"""
r = _RANK[role]
return -1 if r is None else r
def owner_admin(role):
return role in (OWNER, ADMIN)
def given_by(role, own):
""""role <= own" (§3.2, R8.2 step 1): owner and auditor only by an owner; others by an owner or admin of at
least that rank."""
if role in (OWNER, AUDITOR):
return own == OWNER
return owner_admin(own) and _rank_cmp(own) >= _rank_cmp(role)
def is_demotion(frm, to):
a, b = rank(frm), rank(to)
if a is not None and b is not None:
return b < a
if a is not None:
return True
return False
def _sig_key_checked(b):
if type(b) is not bytes:
_bad("key not bytes")
k = sig_key(b)
if isinstance(k, str):
_bad("signature key refused: " + k)
return b
def _kem_key_checked(b):
if type(b) is not bytes:
_bad("receipt or readers key not bytes")
why = kem_key(b)
if why:
_bad("receipt or readers key refused: " + why)
return b
def _u16s(a):
out = []
for v in a:
if not _is_u64(v) or v > 0xFFFF:
_bad("not a 16-bit number")
out.append(v)
return out
class KeyProof:
"""A device's public keys with their proofs of possession (R2.2, R8.6, R10b.2)."""
__slots__ = ("account", "k1", "k2", "k3", "k4", "k14", "stmt", "sigs")
def signing_keys(self):
v = [self.k1, self.k2]
if self.k4 is not None:
v.append(self.k4)
if self.k14 is not None:
v.append(self.k14)
return v
def all_keys(self):
return self.signing_keys() + [self.k3]
def keyproof_from(v):
r = Rd.typed(v, "key-proof")
stmt = r.get("stmt")
s = Rd.typed(stmt, "keys")
p = KeyProof()
p.account = s.hash("acct")
p.k1 = _sig_key_checked(s.bytes("k1"))
p.k2 = _sig_key_checked(s.bytes("k2"))
p.k3 = _kem_key_checked(s.bytes("k3"))
p.k4 = _sig_key_checked(s.opt("k4")) if "k4" in stmt else None
p.k14 = _sig_key_checked(s.opt("k14")) if "k14" in stmt else None
suites = _u16s(s.array("suites"))
formats = _u16s(s.array("formats"))
s.done()
# (R10b.4) this verifier knows suite 1 and format 1 only
if suites != [SUITE] or formats != [FORMAT]:
_bad("support for suites %r and formats %r: suite 1 and format 1 only" % (suites, formats))
sigs = r.array("sigs")
if not all(type(x) is bytes for x in sigs):
_bad("key-proof signatures not bytes")
r.done()
p.stmt = cenc(stmt)
p.sigs = sigs
if len(sigs) != len(p.signing_keys()):
_bad("one proof-of-possession signature per signing key")
return p
def verify_keys(p):
ks = p.signing_keys()
return len(ks) == len(p.sigs) and all(hybrid_verify(k, L_SUPPORT, p.stmt, s) for k, s in zip(ks, p.sigs))
REASONS = ("new-place", "membership", "rotation", "recovery", "switch-over", "freeze-end", "scheduled")
class Gen:
"""A generation G(place, n) as the log carries it (R5.1)."""
__slots__ = ("place", "n", "prev", "reason", "recipients", "commit", "readers", "hash")
def recipient_ids(self):
return [d for d, _ in self.recipients]
def gen_from(v):
r = Rd.typed(v, "generation")
g = Gen()
g.place = r.u64("place")
g.n = r.u64("n")
prev = r.get("prev")
if prev is None:
g.prev = None
else:
g.prev = _hash_of(prev)
if g.prev is None:
_bad("prev not a hash")
g.reason = r.text("reason")
if g.reason not in REASONS:
_bad("reason %r" % g.reason)
g.recipients = []
for x in r.array("recipients"):
if type(x) is not list or len(x) != 2:
_bad("recipient not [device, box]")
d, bx = x
if not _is_u64(d):
_bad("recipient id")
if g.recipients and g.recipients[-1][0] >= d:
_bad("recipients not strictly increasing")
if type(bx) is not bytes:
_bad("box not bytes")
why = sealed_ok(bx)
if why:
_bad("sealed box refused: " + why)
g.recipients.append((d, bx))
g.commit = r.hash("commit")
g.readers = [_kem_key_checked(k) for k in r.array("readers")]
r.done()
if len(g.readers) != 8:
_bad("a generation lists exactly eight readers keys (R5.1)")
if (g.n == 0) != (g.prev is None):
_bad("generation 0 has no prev, every later one has")
g.hash = sha256(cenc(v))
return g
def gens_from(a):
gens = [gen_from(x) for x in a]
for x, y in zip(gens, gens[1:]):
if x.place >= y.place:
_bad("generations not in strictly increasing place order")
return gens
class Transcript:
"""A console join as the proposal carries it (R8.2 steps 2-5; WP6): moxivue-proto's pairing transcript
{sid, ya, a: {k1, k2, k3, support, commit}, yb, b: {account, inviter, role, reach, person, nonce}} and the joiner's
revealed nA. The keys, role, reach and person the join admits are read from it (F1); its hash and words are
moxivue-proto's derivation (twin: tools/moxivue_pairing.py)."""
__slots__ = ("account", "inviter", "joiner", "role", "reach", "person", "automation", "pairing", "nonce_a",
"nonce_b")
def hash(self):
return labelled_hash(L_PAIRING_TRANSCRIPT, cenc(self.pairing))
def words(self):
"""the three word indices (R8.2 steps 4, 6): 33 bits of H(words, ["words", transcript hash, nA, nB])"""
d = labelled_hash(L_WORDS, cenc(["words", self.hash(), self.nonce_a, self.nonce_b]))
top = int.from_bytes(d[:5], "big") >> 7
return [(top >> 22) & 0x7FF, (top >> 11) & 0x7FF, top & 0x7FF]
def _fields(v, keys, what):
"""exactly these keys (moxivue-proto pairing.rs fields)"""
if type(v) is not dict or len(v) != len(keys) or any(k not in v for k in keys):
_bad(what)
return [v[k] for k in keys]
def _fixed(v, n, what):
if type(v) is not bytes or len(v) != n:
_bad(what)
return v
def _bounded(v, mx, what):
if type(v) is not bytes or not 1 <= len(v) <= mx:
_bad(what)
return v
def transcript_from(v):
r = Rd.typed(v, "console-join")
pv = r.get("pairing")
nonce_a = _fixed(r.bytes("nonce_a"), 32, "nA is 32 bytes")
r.done()
sid, ya, ada, yb, adb = _fields(pv, ["sid", "ya", "a", "yb", "b"], "transcript")
if type(sid) is not bytes or not 16 <= len(sid) <= 64:
_bad("sid")
_fixed(ya, 32, "ya")
_fixed(yb, 32, "yb")
k1, k2, k3, support, commit = _fields(ada, ["k1", "k2", "k3", "support", "commit"], "ADa")
for k in (k1, k2):
_sig_key_checked(k)
_kem_key_checked(k3)
_bounded(support, 16 * 1024, "support")
_fixed(commit, 32, "commit")
acct, inviter, role, reach, person, nonce_b = _fields(
adb, ["account", "inviter", "role", "reach", "person", "nonce"], "ADb")
_bounded(acct, 128, "account")
_sig_key_checked(inviter)
if type(role) is not str or not 1 <= len(role.encode("utf-8")) <= 32:
_bad("role")
_fixed(nonce_b, 32, "nonce")
# nA must open the joiner's commitment (R8.2 step 4)
if labelled_hash(L_WORDS, cenc(["commit", nonce_a])) != commit:
_bad("nA does not open the joiner's commitment (R8.2 step 4)")
t = Transcript()
if len(acct) != 32:
_bad("the offer's account is not a fingerprint")
t.account = acct
if role not in ROLES:
_bad("offered role %r" % role)
t.role = role
if not _is_u64(reach):
_bad("offered reach is not a place")
t.reach = reach
pr = Rd(person, "person")
t.person = pr.text("of")
if t.person not in ("new", "mine"):
_bad("person %r" % t.person)
t.automation = pr.bool("automation")
pr.done()
# the joiner's key statement from its keys and support (moxivue-log KeyProof::with_support)
try:
sv = cdec(support)
except CanonError as e:
_bad("support: " + e.args[0])
sr = Rd.typed(sv, "support")
suites = sr.array("suites")
formats = sr.array("formats")
sigs = sr.array("sigs")
sr.done()
stmt = {"t": "keys", "acct": acct, "k1": k1, "k2": k2, "k3": k3, "suites": suites, "formats": formats}
t.joiner = keyproof_from({"t": "key-proof", "stmt": stmt, "sigs": sigs})
t.inviter = inviter
t.pairing = pv
t.nonce_a = nonce_a
t.nonce_b = nonce_b
return t
# the capabilities a settings entry may declare in clear, with their value types (0 on/off, 1 days, 2 list hash)
CAPS = {"ssh.host": 0, "users": 0, "terminal": 0, "forwards": 0, "remote": 0, "sudo": 0, "retention": 1, "dispatcher": 2,
"witnesses": 2, "contacts": 2}
class Settings:
__slots__ = ("place", "machine", "version", "caps")
def scope(self):
"""what versions count per: the place, or the one machine (WP7, R6.14); ids are log positions, never equal"""
return self.machine if self.machine is not None else self.place
def settings_from(v):
r = Rd.typed(v, "settings")
st = Settings()
st.place = r.u64("place")
st.machine = r.u64("machine") if "machine" in v else None
st.version = r.u64("version")
cm = r.get("caps")
if type(cm) is not dict:
_bad("caps not a map")
st.caps = []
for k in sorted(cm, key=_key_order):
x = cm[k]
ty = CAPS.get(k)
if ty is None:
_bad("unknown capability %r" % k)
if ty == 0 and type(x) is bool:
st.caps.append((k, ("on", x)))
elif ty == 1:
if not _is_u64(x):
_bad("days")
st.caps.append((k, ("days", x)))
elif ty == 2:
if _hash_of(x) is None:
_bad("list hash")
st.caps.append((k, ("list", x)))
else:
_bad("capability %r of the wrong type" % k)
r.hash("doc_hash")
r.bytes("doc")
r.bytes("keys")
r.done()
return st
class Change:
"""A sensitive change (R3.5): kind and fields."""
__slots__ = ("kind", "f")
def __init__(self, kind, **f):
self.kind = kind
self.f = f
def target(self):
if self.kind in ("remove", "role", "reach", "move", "person"):
return self.f["target"]
return None
def change_from(v):
r = Rd.typed(v, "change")
k = r.text("kind")
if k == "join":
c = Change(k, t=transcript_from(r.get("transcript")))
elif k == "remove":
c = Change(k, target=r.u64("target"))
elif k == "role":
c = Change(k, target=r.u64("target"), role=r.role("role"))
elif k == "reach":
c = Change(k, target=r.u64("target"), reach=r.u64("reach"))
elif k == "move":
c = Change(k, target=r.u64("target"), place=r.u64("place"))
elif k == "person":
c = Change(k, target=r.u64("target"), person=r.u64("person"))
elif k == "history":
c = Change(k, person=r.u64("person"), consoles=r.devices("consoles"), sealed=r.bytes("sealed"))
elif k == "recovery-replace":
c = Change(k, key=_sig_key_checked(r.bytes("key")), holder=r.u64("holder"))
elif k == "settings":
c = Change(k, st=settings_from(r.get("settings")))
elif k == "delete-records":
c = Change(k, place=r.u64("place"), before=r.u64("before"))
elif k == "switch-over":
suite = r.u64("suite")
if suite > 0xFFFF:
_bad("suite")
c = Change(k, suite=suite, emergency=r.bool("emergency"), exclude=r.devices("exclude"))
elif k == "escalate":
c = Change(k, proposal=r.hash("proposal"), exclude=r.devices("exclude"))
elif k == "delete-account":
c = Change(k)
else:
_bad("unknown change %r" % k)
r.done()
return c
class CancelList:
"""A witness's signed list of the cancels it holds for a recovery request (R4.11, R9.5)."""
__slots__ = ("witness", "request", "time", "cancels", "proof", "signed_part")
def cancel_list_from(v):
r = Rd.typed(v, "cancel-list-signed")
lv = r.get("list")
lr = Rd.typed(lv, "cancel-list")
c = CancelList()
c.witness = lr.u64("witness")
if c.witness >= 2:
_bad("witness")
c.request = lr.hash("request")
c.time = lr.u64("time")
c.cancels = lr.array("cancels")
if not all(type(x) is bytes for x in c.cancels):
_bad("cancel not bytes")
lr.done()
c.proof = r.bytes("proof")
r.done()
c.signed_part = lv
return c
def _name(s):
b = s.encode("utf-8")
if not b or len(b) > 64 or any(ord(ch) < 0x20 or 0x7F <= ord(ch) <= 0x9F for ch in s):
_bad("a name is 1-64 bytes without control characters")
return s
def _machine_keys(p):
if p.k4 is None or p.k14 is None:
_bad("a machine carries its record key K4 and executor key K14 (R2.2)")
return p
def body_from(kind, v):
r = Rd.typed(v, "body")
b = {}
if kind == "genesis":
b["name"] = _name(r.text("name"))
b["person"] = _name(r.text("person"))
b["console"] = keyproof_from(r.get("console"))
b["recovery"] = _sig_key_checked(r.bytes("recovery"))
b["gen"] = gen_from(r.get("gen"))
elif kind == "propose":
b["change"] = change_from(r.get("change"))
elif kind in ("confirm", "refuse", "withdraw"):
b["proposal"] = r.hash("proposal")
elif kind == "finish":
b["proposal"] = r.hash("proposal")
b["gens"] = gens_from(r.array("gens"))
elif kind == "machine-join":
b["place"] = r.u64("place")
r.hash("sas")
b["machine"] = _machine_keys(keyproof_from(r.get("machine")))
r.bytes("docs")
elif kind == "token-issue":
b["place"] = r.u64("place")
b["expiry"] = r.u64("expiry")
b["tokens"] = [_sig_key_checked(k) for k in r.array("tokens")]
elif kind == "token-revoke":
b["issue"] = r.u64("issue")
b["tokens"] = [_sig_key_checked(k) for k in r.array("tokens")]
elif kind == "token-join":
b["issue"] = r.u64("issue")
b["token"] = _sig_key_checked(r.bytes("token"))
b["machine"] = _machine_keys(keyproof_from(r.get("machine")))
r.bytes("nonces")
elif kind == "leave":
pass
elif kind == "place-new":
b["parent"] = r.u64("parent")
b["name"] = _name(r.text("name"))
b["gen"] = gen_from(r.get("gen"))
elif kind == "place-name":
b["place"] = r.u64("place")
b["name"] = _name(r.text("name"))
elif kind == "person-name":
b["person"] = r.u64("person")
b["name"] = _name(r.text("name"))
elif kind == "generation":
b["gen"] = gen_from(r.get("gen"))
elif kind == "rotate":
b["keys"] = keyproof_from(r.get("keys"))
b["gens"] = gens_from(r.array("gens"))
elif kind == "support":
suites = _u16s(r.array("suites"))
formats = _u16s(r.array("formats"))
if suites != [SUITE] or formats != [FORMAT]:
_bad("suite 1, format 1 only (R10b.4)")
elif kind == "settings":
b["st"] = settings_from(r.get("settings"))
elif kind == "ssh-grant":
b["holder"] = r.u64("holder")
b["machine"] = r.u64("machine")
r.bytes("key")
b["expiry"] = r.u64("expiry")
b["unattended"] = r.bool("unattended")
elif kind == "recovery-request":
b["console"] = keyproof_from(r.get("console"))
if b["console"].k4 is not None or b["console"].k14 is not None:
_bad("a console has no record or executor key")
b["named"] = r.devices("named")
snap = []
for x in r.array("snapshot"):
sr = Rd.typed(x, "snap")
it = (sr.u64("device"), sr.hash("k1"), sr.u64("reach"))
sr.done()
if snap and snap[-1][0] >= it[0]:
_bad("snapshot not strictly increasing")
snap.append(it)
b["snapshot"] = snap
elif kind == "cancel":
b["request"] = r.hash("request")
elif kind == "recovery-complete":
b["request"] = r.hash("request")
b["start"] = r.base("start")
b["lists"] = [cancel_list_from(x) for x in r.array("lists")]
b["key"] = _sig_key_checked(r.bytes("key"))
b["gens"] = gens_from(r.array("gens"))
else:
_bad("unknown entry kind %r" % kind)
r.done()
return b
class Entry:
__slots__ = ("account", "id", "signer", "base", "kind", "b", "hash")
def signer_id(self):
return self.signer if type(self.signer) is int else None
def signatures(self):
"""(whose key, label) of each signature, in order (moxivue-log entry.rs Entry::signatures)"""
k = self.kind
if k == "genesis":
return [("carried", L_ENTRY)]
if k == "propose":
return [("k1", L_PROPOSAL)]
if k == "confirm":
return [("k1", L_CONFIRMATION)]
if k == "cancel":
return [("k1", L_CANCEL)]
if k == "support":
return [("k1", L_SUPPORT)]
if k == "rotate":
v = [("k1", L_ROTATION), ("carried", L_ROTATION)]
if self.b["keys"].k14 is not None:
v.append(("k14", L_ROTATION))
return v
if k == "token-join":
return [("carried", L_ENTRY), ("token", L_TOKEN_JOIN)]
if k == "recovery-request":
return [("recovery", L_RECOVERY_REQUEST), ("carried", L_RECOVERY_REQUEST)]
return [("k1", L_ENTRY)]
def entry_from(eb):
try:
v = cdec(eb)
except CanonError as e:
_bad("encoding: " + e.args[0])
r = Rd.typed(v, "entry")
if r.u64("v") != FORMAT:
_bad("unknown format version (R2.6)")
e = Entry()
e.kind = r.text("kind")
e.id = r.bytes("id")
if len(e.id) != 16:
_bad("id is 16 bytes")
by = r.get("by")
if by == "new" and type(by) is str:
e.signer = "new"
elif by == "recovery" and type(by) is str:
e.signer = "recovery"
elif _is_u64(by):
e.signer = by
else:
_bad("signer")
acct = r.opt("acct")
if acct is not None or "acct" in v:
if _hash_of(acct) is None:
_bad("acct")
e.account = acct
bv = r.opt("base")
e.base = None if "base" not in v else base_from(bv)
e.b = body_from(e.kind, r.get("body"))
r.done()
genesis = e.kind == "genesis"
if genesis != (e.account is None) or genesis != (e.base is None):
_bad("only the genesis has no account and no base")
wants_new = e.kind in ("genesis", "token-join")
wants_recovery = e.kind == "recovery-request"
if e.signer == "new":
fits = wants_new
elif e.signer == "recovery":
fits = wants_recovery
else:
fits = not wants_new and not wants_recovery
if not fits:
_bad("signer does not fit the entry kind")
e.hash = sha256(eb)
return e
class Signed:
__slots__ = ("entry", "bytes", "sigs")
def signed_from(b):
try:
v = cdec(b)
except CanonError as e:
_bad("encoding: " + e.args[0])
r = Rd.typed(v, "signed")
eb = r.bytes("e")
sigs = r.array("s")
if not all(type(x) is bytes for x in sigs):
_bad("signature not bytes")
r.done()
s = Signed()
s.entry = entry_from(eb)
s.bytes = eb
s.sigs = sigs
if len(sigs) != len(s.entry.signatures()):
_bad("wrong number of signatures for the entry kind")
return s
def leaf_from(b):
try:
v = cdec(b)
except CanonError as e:
_bad("leaf encoding: " + e.args[0])
if type(v) is not list or len(v) != 4 or v[0] != "leaf" or type(v[0]) is not str or type(v[3]) is not bytes:
_bad("not a leaf")
if not _is_u64(v[1]):
_bad("leaf seq")
if _hash_of(v[2]) is None:
_bad("leaf prev")
return v[1], v[2], v[3]
def leaf_hash(b):
return sha256(b"\x00" + b)
def node_hash(l, r):
return sha256(b"\x01" + l + r)
class Tree:
"""RFC 6962 Merkle tree of the leaves, with the root of every prefix."""
def __init__(self):
self.leaves = []
self.roots = [sha256(b"")]
self.frontier = []
def push(self, h):
self.leaves.append(h)
cur, height = h, 0
while self.frontier and self.frontier[-1][1] == height:
top, _ = self.frontier.pop()
cur, height = node_hash(top, cur), height + 1
self.frontier.append((cur, height))
it = reversed(self.frontier)
r = next(it)[0]
for x, _ in it:
r = node_hash(x, r)
self.roots.append(r)
def export_from(b):
try:
v = cdec(b)
except CanonError as e:
_bad("export encoding: " + e.args[0])
r = Rd.typed(v, "log-export")
if r.u64("v") != 1:
_bad("unknown export version")
leaves = r.array("leaves")
if not all(type(x) is bytes for x in leaves):
_bad("leaf not bytes")
cps = []
for c in r.array("checkpoints"):
cr = Rd.typed(c, "checkpoint")
cps.append((cr.base("base"), cr.bytes("proof")))
cr.done()
r.done()
return leaves, cps
# ======================================================================== the rules (crates/moxivue-log: model, core,
# state, may, apply). Times are witness time (seconds) from the base checkpoints entries name (R4.4).
HOUR = 3600
DAY = 24 * HOUR
TOLERANCE = 5 * 60 # the most two witnesses' stamps of one checkpoint may differ (R4.4)
LAPSE = DAY # a proposal lapses a day after its base (R3.9)
SWITCHOVER_LAPSE = 30 * DAY # a normal switch-over (R10b.9)
DELAY_D = 7 * DAY # recovery delay D (R9.3)
DELAY_D2 = 14 * DAY # second recovery delay D2 (R9.4)
TOKEN_MAX = DAY # install tokens expire within 24 hours (R8.10)
SSH_GRANT_MAX = 30 * DAY # SSH key grants (R6.14)
SETUP_DEVICES, SETUP_OWNERS = 3, 2 # leaving set-up (R3.12, 4.2-10)
FLOOR_DEVICES, FLOOR_OWNERS = 2, 1 # the floor D1 (R3.12)
MAX_ENTRIES = 50000 # the longest log the verifier takes (build plan assumption 5)
ACCOUNT = 0 # the whole account as a place
CONSOLE, MACHINE = "console", "machine"
WAITING, ACTIVE, REMOVED, EXCLUDED = "waiting", "active", "removed", "excluded"
class Refusal(Exception):
def __init__(self, rule, why):
Exception.__init__(self, rule, why)
self.rule = rule
self.why = why
class Fail(Exception):
"""A verification failure: the log does not verify from here."""
def need(ok, rule, why):
if not ok:
raise Refusal(rule, why() if callable(why) else why)
def dev_name(d):
return "device %d" % d
def place_name(p):
return "the whole account" if p == ACCOUNT else "place %d" % p
# What a device is at one log size.
DV = namedtuple("DV", "kind role reach person automation status")
def dv_console(d):
return d.kind == CONSOLE
def dv_owner_admin(d):
return d.role is not None and owner_admin(d.role)
class Device:
__slots__ = ("id", "view", "k1", "k2", "k3", "k4", "k14", "old_k1", "recovery_epoch", "hist")
def __init__(self, id, view, k1, k2, k3, k4, k14, recovery_epoch, size):
self.id = id
self.view = view
self.k1, self.k2, self.k3, self.k4, self.k14 = k1, k2, k3, k4, k14
self.old_k1 = []
self.recovery_epoch = recovery_epoch
self.hist = [(size, view)]
def at(self, size):
for s, v in reversed(self.hist):
if s <= size:
return v
return None
PENDING, FINISHED, REFUSED_P, WITHDRAWN, VOID = "pending", "finished", "refused", "withdrawn", "void"
R_PENDING, R_ENDED, R_VOID, R_COMPLETED = "pending", "ended", "void", "completed"
class Proposal:
__slots__ = ("hash", "seq", "proposer", "change", "base", "status", "confirmations", "confirmed", "ended",
"freezes")
def deadline(self):
c = self.change
lapse = SWITCHOVER_LAPSE if c.kind == "switch-over" and not c.f["emergency"] else LAPSE
return base_later(self.base) + lapse
def freeze_target(c, frm):
"""R3.11: the device a counting confirmation of this change freezes"""
if c.kind in ("remove", "reach", "move"):
return c.f["target"]
if c.kind == "role":
if frm is None:
return None
return c.f["target"] if is_demotion(frm, c.f["role"]) else None
return None
class Request:
__slots__ = ("hash", "seq", "console", "named", "snapshot", "holder", "key", "status", "extended",
"named_cancels")
class Recovery:
__slots__ = ("key", "holder", "suspended", "epoch")
def __init__(self, key, holder, suspended, epoch):
self.key, self.holder, self.suspended, self.epoch = key, holder, suspended, epoch
class Token:
__slots__ = ("key", "issue", "place", "expiry", "revoked", "used")
class Place:
__slots__ = ("id", "parent", "name", "since")
def __init__(self, id, parent, name, since):
self.id, self.parent, self.name, self.since = id, parent, name, since
def opt_lt(a, b):
"""a < b for optional numbers, with nothing below every number (Rust's Option order)"""
if a is None:
return b is not None
if b is None:
return False
return a < b
class Touch:
__slots__ = ("devices", "gens", "proposal")
def __init__(self):
self.devices = []
self.gens = []
self.proposal = None
class State:
def __init__(self):
self.account = bytes(32)
self.name = ""
self.tree = Tree()
self.devices = {}
self.device_ids = [] # sorted
self.persons = {}
self.places = {}
self.gens = {} # place -> (n, hash, seq)
self.proposals = {} # hash -> Proposal
self.freezing = {} # device -> proposals that freeze it once confirmed (R3.11)
self.removals = [] # proposals of removals
self.removals_of = {} # device -> proposals of its removal
self.requests = {} # hash -> Request
self.recovery = None
self.tokens = []
self.settings = {} # place -> [version, {cap: value}]
self.keys_seen = set()
self.ids = set()
self.touches = []
self.now_sizes = [0]
self.now_times = [0]
self.floor_since = None
self.single_exception = False
self.deleted = False
self.checkpoints = {} # size -> [Base]
self.gen_list = {} # place -> [(n, position of the entry carrying it)], chain order (records)
self.k4_hist = {} # machine -> [(log size from which it holds, K4)] (records, R6.8)
self.refused = [] # (position, Refusal)
def len(self):
return len(self.tree.leaves)
def device(self, i):
return self.devices.get(i)
def add_device(self, d):
if d.id not in self.devices:
bisect.insort(self.device_ids, d.id)
self.devices[d.id] = d
def add_proposal(self, p):
self.proposals[p.hash] = p
if p.freezes is not None:
self.freezing.setdefault(p.freezes, []).append(p)
if p.change.kind == "remove":
self.removals.append(p)
self.removals_of.setdefault(p.change.f["target"], []).append(p)
def cosigned(self):
return max(self.checkpoints) if self.checkpoints else 0
def now(self):
return View(self, self.len())
def at(self, size):
return View(self, size)
def set_view(self, i, size, view):
d = self.devices[i]
d.view = view
if d.hist and d.hist[-1][0] == size:
d.hist[-1] = (size, view)
else:
d.hist.append((size, view))
def known_base(self, b):
return b in self.checkpoints.get(b.size, ())
def recovery_epoch(self):
return self.recovery.epoch if self.recovery is not None else 0
class View:
"""The state as it was after the first `at` entries."""
__slots__ = ("s", "at")
def __init__(self, s, at):
self.s = s
self.at = at
def device(self, i):
d = self.s.devices.get(i)
return d.at(self.at) if d is not None else None
def devices(self):
out = []
for i in self.s.device_ids:
v = self.s.devices[i].at(self.at)
if v is not None:
out.append((i, v))
return out
def time(self):
k = bisect.bisect_right(self.s.now_sizes, self.at) - 1
return self.s.now_times[k] if k >= 0 else 0
def place_exists(self, p):
pl = self.s.places.get(p)
return pl is not None and pl.since <= self.at
def covers(self, reach, place):
cur = place
for _ in range(4):
if cur is None:
return False
if cur == reach:
return self.place_exists(cur)
pl = self.s.places.get(cur)
cur = pl.parent if pl is not None else None
return False
def pending(self, p):
return p.seq < self.at and (p.ended is None or self.at < p.ended) and self.time() <= p.deadline()
def confirmed_pending(self, p):
return p.confirmed is not None and p.confirmed <= self.at and self.pending(p)
def frozen(self, i):
return any(self.confirmed_pending(p) for p in self.s.freezing.get(i, ()))
def removal_target(self, i):
return any(self.confirmed_pending(p) for p in self.s.removals_of.get(i, ()))
def acting(self, i):
d = self.device(i)
if d is not None and d.status == ACTIVE and not self.frozen(i):
return d
return None
def shape(self):
n = o = 0
for _, d in self.devices():
if dv_console(d) and dv_owner_admin(d) and d.reach == ACCOUNT and not d.automation and d.status == ACTIVE:
n += 1
if d.role == OWNER:
o += 1
return n, o
def setup(self):
n, o = self.shape()
return not (n >= SETUP_DEVICES and o >= SETUP_OWNERS)
def owner_admin_count(self):
return sum(1 for _, d in self.devices() if dv_console(d) and dv_owner_admin(d) and d.status == ACTIVE)
def recipients(self, place):
return [i for i, d in self.devices()
if dv_console(d) and d.status == ACTIVE and self.covers(d.reach, place) and not self.frozen(i)]
def places(self):
return sorted(p.id for p in self.s.places.values() if p.since <= self.at)
def covered(self, reach):
return [p for p in self.places() if self.covers(reach, p)]
def place_depth(s, p):
pl = s.places.get(p)
if pl is None:
return None
if pl.parent is None:
return 0
d = place_depth(s, pl.parent)
return 1 + (d if d is not None else 0)
# ------------------------------------------------------------------------ core decisions (moxivue-log core.rs)
def core_standing(status):
if status == WAITING:
raise Refusal("R3.13", "a waiting device has no right")
if status != ACTIVE:
raise Refusal("§3.1", "not an active device")
def core_confirm(m, frozen, is_proposer, is_target, removal_target, covers, kind, special):
"""R3.1-R3.3 (and R9.1, R6.13, R10b.10): an eligible confirmer; special is named / grantee / excluded."""
core_standing(m.status)
if m.kind != CONSOLE or m.role is None:
raise Refusal("R3.1", "only consoles confirm")
need(owner_admin(m.role), "R3.1", "only owner and admin consoles confirm (never an auditor, AUD-1)")
need(not m.automation, "R3.1", "an automation device never confirms (W-35)")
need(not frozen, "R3.1", "a frozen device confirms nothing (R3.11)")
need(not is_proposer, "R3.3", "the proposer never confirms its own proposal")
need(not is_target, "R3.2", "the target never confirms its own change")
need(not removal_target, "R3.1", "the target of a pending removal confirms nothing")
need(covers, "R3.1", "its reach does not cover the change")
if kind == "recovery-replace":
need(not special, "R9.1", "named in a pending recovery request")
elif kind == "history":
need(not special, "R6.13", "of the grantee's person")
elif kind == "switch-over":
need(not special, "R10b.10", "an excluded device never confirms its own exclusion")
def floor_ok(reached, n, o):
return not reached or (n >= FLOOR_DEVICES and o >= FLOOR_OWNERS)
def conflicts(e_proposal, e_devices, e_gens, proposal, target, places):
return e_proposal == proposal or (target is not None and target in e_devices) or any(g in places for g in e_gens)
def cancels_end(facts):
"""R9.4: (device, named, whole) facts end the request for good: a not-named whole-account cancel, or two
different named devices."""
first_named = None
for d, named, whole in facts:
if not named and whole:
return True
if named:
if first_named is None:
first_named = d
elif first_named != d:
return True
return False
# ------------------------------------------------------------------------ may() (moxivue-log may.rs), log permissions
def may(member, perm, target, v):
"""Raises Refusal unless `member` may do `perm` on `target` in view `v`. perm: ("sign", leaving), ("manage",),
("generation",), ("sshgrant",), ("cancel",), ("refuse",), ("confirm",), ("propose", what, arg).
target: None, ("device", id), ("place", id), ("proposal", p), ("request", r)."""
m = v.device(member)
if m is None:
raise Refusal("§3.2", "%s is not in the log" % dev_name(member))
frozen = v.frozen(member)
try:
core_standing(m.status)
except Refusal as r:
raise Refusal(r.rule, "%s: %s" % (dev_name(member), r.why))
k = perm[0]
if k == "sign":
if frozen and not perm[1]:
raise Refusal("R3.11", "%s: frozen: only its own leaving" % dev_name(member))
return
if k == "manage":
need(not frozen, "R3.11", lambda: "%s is frozen" % dev_name(member))
need(dv_owner_admin(m), "§3.2", "only an owner or admin")
return place_in_reach(m, target, v)
if k == "generation":
need(not frozen, "R5.2", lambda: "%s is frozen: it signs no generation" % dev_name(member))
need(dv_console(m) and dv_owner_admin(m), "R5.2", "generations are signed by an owner/admin console")
return place_in_reach(m, target, v)
if k == "sshgrant":
need(not frozen, "R3.11", lambda: "%s is frozen" % dev_name(member))
need(m.role == OWNER, "R6.14", "SSH key grants are owner-signed")
return machine_in_reach(m, target, v)
if k == "cancel":
if target is None or target[0] != "request":
raise Refusal("R9.4", "a cancel names a request")
need(not frozen, "R3.11", lambda: "%s is frozen" % dev_name(member))
need(any(x[0] == member for x in target[1].snapshot), "R9.4",
lambda: "%s is not an owner/admin device of the request's snapshot (CancelOnlyPre)" % dev_name(member))
return
if k == "refuse":
if target is None or target[0] != "proposal":
raise Refusal("R3.13", "a refusal names a proposal")
p = target[1]
need(not frozen, "R3.11", lambda: "%s is frozen" % dev_name(member))
need(dv_owner_admin(m), "R3.13", "only owner/admin devices refuse")
inviter = v.device(p.proposer)
counts = (member == p.proposer or (m.person is not None and inviter is not None and inviter.person == m.person)
or m.role == OWNER)
need(counts, "R3.13", "a refusal counts only from the inviting device, its person's devices or an owner device"
" (W-25)")
return
if k == "confirm":
if target is None or target[0] != "proposal":
raise Refusal("R3.1", "a confirmation names a proposal")
return eligible(member, m, frozen, target[1], v)
if k == "propose":
return propose_rule(member, m, frozen, perm[1], perm[2] if len(perm) > 2 else None, target, v)
raise Refusal("§3.2", "unknown permission")
def place_in_reach(m, target, v):
if target is None or target[0] != "place":
raise Refusal("§3.2", "no place named")
p = target[1]
need(v.place_exists(p), "§3.1", lambda: "%s does not exist" % place_name(p))
need(v.covers(m.reach, p), "§3.2", lambda: "%s is outside the reach" % place_name(p))
def machine_in_reach(m, target, v):
if target is None or target[0] != "device":
raise Refusal("R10.1", "a machine must be named")
t = target[1]
tv = v.device(t)
if tv is None:
raise Refusal("R10.1", "%s is not in the log" % dev_name(t))
need(tv.kind == MACHINE and tv.status == ACTIVE, "R10.1", lambda: "%s is not an active machine" % dev_name(t))
need(v.covers(m.reach, tv.reach), "R10.2", lambda: "%s is outside the reach" % dev_name(t))
def eligible(member, m, frozen, p, v):
c = p.change
place = change_place(p, v)
special = False
if c.kind == "recovery-replace":
special = any(r.status == R_PENDING and member in r.named for r in v.s.requests.values())
elif c.kind == "history":
special = m.person == c.f["person"]
elif c.kind in ("switch-over", "escalate"):
special = member in c.f["exclude"]
kind = {"recovery-replace": "recovery-replace", "history": "history", "switch-over": "switch-over",
"escalate": "switch-over"}.get(c.kind, "other")
try:
core_confirm(m, frozen, member == p.proposer, c.target() == member, v.removal_target(member),
v.covers(m.reach, place), kind, special)
except Refusal as r:
raise Refusal(r.rule, "%s: %s" % (dev_name(member), r.why))
def change_place(p, v):
c = p.change
if c.kind == "join":
return c.f["t"].reach
if c.kind in ("remove", "role", "person"):
d = v.device(c.f["target"])
return d.reach if d is not None else ACCOUNT
if c.kind in ("reach", "move"):
d = v.device(c.f["target"])
old = d.reach if d is not None else ACCOUNT
return common_place(old, c.f["reach"] if c.kind == "reach" else c.f["place"], v)
if c.kind == "settings":
return c.f["st"].place
if c.kind == "delete-records":
return c.f["place"]
return ACCOUNT
def common_place(a, b, v):
cur = a
for _ in range(4):
if cur is None:
break
if v.covers(cur, b):
return cur
pl = v.s.places.get(cur)
cur = pl.parent if pl is not None else None
return ACCOUNT
def propose_rule(member, m, frozen, what, arg, target, v):
need(not frozen, "R3.11", lambda: "%s is frozen" % dev_name(member))
need(dv_console(m), "§3.2", "only consoles propose")
role = m.role
need(owner_admin(role), "§3.2", lambda: "a %s proposes no change" % role)
owner = role == OWNER
tdev = None
if target is not None and target[0] == "device":
d = v.device(target[1])
if d is None:
raise Refusal("§3.2", "%s is not in the log" % dev_name(target[1]))
tdev = (target[1], d)
def target_dev():
if tdev is None:
raise Refusal("§3.2", "no device named")
return tdev
if what == "join":
r, reach = arg
need(given_by(r, role), "R8.2", lambda: "a %s may not invite a %s (role ≤ own; owner and auditor only by an "
"owner, AUD-1)" % (role, r))
need(v.place_exists(reach) and v.covers(m.reach, reach), "R8.2", "the reach must be within the inviter's")
need(not v.setup() or owner_admin(r), "R3.12", "in set-up only owner and admin consoles join")
elif what == "remove":
t, tv = target_dev()
need(t != member, "§3.2", "a device leaves on its own with a leaving entry, not a removal")
need(tv.status == ACTIVE, "§3.2", lambda: "%s is %s" % (dev_name(t), tv.status))
touches(role, tv, v, m)
elif what == "role":
_, tv = target_dev()
need(dv_console(tv) and tv.status == ACTIVE, "§3.2", "a role belongs to an active console")
need(tv.role != arg, "§3.2", "that is its role already")
need(given_by(arg, role), "§3.2", lambda: "a %s may not give the role %s" % (role, arg))
touches(role, tv, v, m)
elif what in ("reach", "move"):
_, tv = target_dev()
want_console = what == "reach"
need(dv_console(tv) == want_console and tv.status == ACTIVE, "§3.2",
"a console's reach (Reach) or a machine's place (Move)")
need(tv.reach != arg, "§3.2", "it is there already")
need(v.place_exists(arg) and v.covers(m.reach, arg), "§3.2", lambda: "%s is outside the reach" % place_name(arg))
touches(role, tv, v, m)
elif what == "person":
_, tv = target_dev()
need(dv_console(tv) and tv.status == ACTIVE, "R3.5", "a person has consoles only")
touches(role, tv, v, m)
elif what == "history":
need(owner, "R6.13", "history grants are an owner's")
epoch = v.s.recovery_epoch()
d = v.s.devices.get(member)
enrolled = d.recovery_epoch if d is not None else epoch
need(enrolled < epoch or epoch == 0, "R6.13",
"signed by an owner device enrolled before the account's most recent recovery")
elif what == "replace-recovery":
need(owner, "R9.1", "an owner proposes the replacement of the recovery identity")
named = any(r.status == R_PENDING and member in r.named for r in v.s.requests.values())
need(not named, "R9.1", "a device named in a pending recovery request does not propose a replacement")
elif what == "settings":
place_in_reach(m, target, v)
elif what in ("settings-owner", "delete-records"):
need(owner, "R3.5", "only an owner proposes this change (§3.2, R6.14, R10.2 b)")
place_in_reach(m, target, v)
elif what in ("switch-over", "delete-account"):
need(owner, "R10b.3", "only an owner proposes this change")
need(m.reach == ACCOUNT, "§3.2", "an account-wide change needs whole-account reach")
else:
raise Refusal("§3.2", "unknown proposal")
def touches(role, tv, v, m):
need(v.covers(m.reach, tv.reach), "§3.2", "the device is outside the reach")
if tv.role is not None:
need(given_by(tv.role, role), "§3.2", lambda: "a %s may not change a %s (only owners touch owners, AUD-1)"
% (role, tv.role))
def is_ok(f, *a):
try:
f(*a)
return True
except Refusal:
return False
# ------------------------------------------------------------------------ applying entries (moxivue-log apply.rs)
class Auth:
__slots__ = ("stale_key", "stale_paper", "proofs_ok", "lists_ok", "cancels")
def __init__(self):
self.stale_key = False
self.stale_paper = False
self.proofs_ok = False
self.lists_ok = []
self.cancels = []
def add_checkpoint(s, base, proof, w):
if abs(base.t1 - base.t2) > TOLERANCE:
raise Fail("checkpoint: stamps %d and %d differ by more than the tolerance (R4.4)" % (base.t1, base.t2))
if base.size == 0 or base.size > s.len():
raise Fail("checkpoint: size %d not in the log of %d entries" % (base.size, s.len()))
if s.tree.roots[base.size] != base.root:
raise Fail("checkpoint: root of size %d differs from the log's: a fork (P18)" % base.size)
if not w.checkpoint(s.account, base, proof):
raise Fail("checkpoint: checkpoint of size %d not cosigned by both witnesses" % base.size)
at = s.checkpoints.setdefault(base.size, [])
if base not in at:
at.append(base)
def carried_keys(e):
k = e.kind
if k == "genesis":
return e.b["console"]
if k == "rotate":
return e.b["keys"]
if k == "token-join":
return e.b["machine"]
if k == "recovery-request":
return e.b["console"]
return None
def authenticate(s, seq, sg, w):
e = sg.entry
auth = Auth()
if seq == 0:
if e.kind != "genesis":
raise Fail("the log does not start with a genesis")
else:
if e.account != s.account:
raise Fail("entry of another account")
b = e.base
if b.size > seq or not s.known_base(b):
raise Fail("checkpoint: base of size %d is not a verified checkpoint of this log before position %d"
% (b.size, seq))
carried = carried_keys(e)
for (role, label), sig in zip(e.signatures(), sg.sigs):
if role == "carried":
ok = hybrid_verify(carried.k1, label, sg.bytes, sig)
elif role == "token":
ok = hybrid_verify(e.b["token"], label, sg.bytes, sig)
elif role == "recovery":
if s.recovery is None:
raise Fail("signature: no recovery identity")
if hybrid_verify(s.recovery.key, label, sg.bytes, sig):
ok = True
elif any(hybrid_verify(k, label, sg.bytes, sig) for k in old_recovery_keys(s)):
auth.stale_paper = True
ok = True
else:
ok = False
else:
i = e.signer_id()
if i is None:
raise Fail("signature: no signing device")
d = s.device(i)
if d is not None:
key = d.k1 if role == "k1" else d.k14
if key is not None and hybrid_verify(key, label, sg.bytes, sig):
ok = True
elif key is not None and role == "k1" and any(hybrid_verify(k, label, sg.bytes, sig) for k in d.old_k1):
auth.stale_key = True
ok = True
else:
ok = False
else:
req = next((r for r in s.requests.values() if r.seq == i), None)
if req is None:
raise Fail("signature: signed by %s, which the log does not know" % dev_name(i))
# the requesting console completes its own recovery before it is a device (R9.5)
ok = role == "k1" and hybrid_verify(req.console.k1, label, sg.bytes, sig)
if not ok:
raise Fail("signature: %s under %s does not verify" % (e.kind, label))
if e.kind == "propose" and e.b["change"].kind == "join":
proofs = [e.b["change"].f["t"].joiner]
elif e.kind == "machine-join":
proofs = [e.b["machine"]]
else:
proofs = [carried] if carried is not None else []
auth.proofs_ok = all(verify_keys(p) for p in proofs)
if e.kind == "recovery-complete":
for l in e.b["lists"]:
auth.lists_ok.append(w.cancel_list(s.account, l) and l.request == e.b["request"])
auth.cancels.append(list_cancels(s, l))
return auth
def old_recovery_keys(s):
cur = s.recovery.key if s.recovery is not None else None
return [r.key for _, r in sorted(s.requests.items()) if r.key != cur]
def list_cancels(s, l):
r = s.requests.get(l.request)
if r is None:
return []
out = []
for c in l.cancels:
try:
sg = signed_from(c)
except DecodeError:
continue
e = sg.entry
if e.kind != "cancel" or e.signer_id() is None:
continue
i = e.signer
if e.b["request"] != l.request or e.account != s.account:
continue
snap = next((x for x in r.snapshot if x[0] == i), None)
if snap is None:
continue
d = s.device(i)
if d is None:
continue
key = next((k for k in [d.k1] + d.old_k1 if sha256(k) == snap[1]), None)
if key is not None and hybrid_verify(key, L_CANCEL, sg.bytes, sg.sigs[0]):
out.append((e.b["request"], i))
return out
class After:
"""The account with the effect of the entry being decided (for the floor and recipient lists)."""
def __init__(self, v):
self.v = v
self.changed = {}
self.unfrozen = None
self.new_place = None
def devices(self):
out = dict(self.v.devices())
out.update(self.changed)
return sorted(out.items())
def covers(self, reach, place):
if self.new_place is not None and self.new_place[0] == place:
p, parent = self.new_place
return p == reach or self.v.covers(reach, parent)
return self.v.covers(reach, place)
def frozen(self, i):
return i != self.unfrozen and self.v.frozen(i)
def shape(self):
n = o = 0
for _, d in self.devices():
if dv_console(d) and dv_owner_admin(d) and d.reach == ACCOUNT and not d.automation and d.status == ACTIVE:
n += 1
o += d.role == OWNER
return n, o
def recipients(self, place):
return [i for i, d in self.devices()
if dv_console(d) and d.status == ACTIVE and self.covers(d.reach, place) and not self.frozen(i)]
def floor_ok(self):
n, o = self.shape()
return floor_ok(self.v.s.floor_since is not None, n, o)
def entry_time(v, e):
return max(base_later(e.base) if e.base is not None else 0, v.time())
def key_halves(k):
body = k[2:]
if k[:2] == TAG_SIG and len(body) > 32:
return [body[:32], body[32:]]
if k[:2] == TAG_KEM and len(body) > 32:
return [body[:-32], body[-32:]]
return [k]
def fresh_keys(s, keys):
"""R2.3, R8.7: no public key half that was ever in the log, and none twice within the entry."""
hs = []
for k in keys:
for half in key_halves(k):
h = sha256(half)
need(h not in s.keys_seen and h not in hs, "R2.3", "a public key that is already in the log")
hs.append(h)
return hs
def check_gen(v, g, reason, want, author_ok):
need(g.reason == reason, "R5.1", lambda: "generation reason %s where %s was due" % (g.reason, reason))
h = v.s.gens.get(g.place)
if h is not None:
need(g.n == h[0] + 1 and g.prev == h[1], "R5.2",
lambda: "generation %d of %s does not extend the chain at %d" % (g.n, place_name(g.place), h[0]))
else:
need(g.n == 0 and g.prev is None, "R5.2", lambda: "%s has no generation chain" % place_name(g.place))
need(author_ok, "R5.2", "generations are signed by an owner/admin console whose reach covers the place")
need(g.recipient_ids() == want, "R5.6", lambda: "recipient list of %s differs from the active, not frozen consoles"
" covering it" % place_name(g.place))
def waits_for_removal(v, place, me):
for p in v.s.removals:
if me is not None and p.hash == me.hash:
continue
if not v.confirmed_pending(p):
continue
if me is not None and me.change.kind == "remove" and not opt_lt(p.confirmed, me.confirmed):
continue
d = v.device(p.change.f["target"])
if d is not None and dv_console(d) and v.covers(d.reach, place):
return True
return False
def carries_pending(v, place):
return any(v.confirmed_pending(p) and place in gen_places(v, p.change) for p in v.s.proposals.values())
def gen_places(v, c):
old = new = None
if c.kind == "join":
new = c.f["t"].reach
elif c.kind in ("remove", "role"):
d = v.device(c.f["target"])
if d is not None and dv_console(d):
old = d.reach
elif c.kind == "reach":
d = v.device(c.f["target"])
if d is not None and dv_console(d):
old, new = d.reach, c.f["reach"]
out = []
for r in (old, new):
if r is None:
continue
for p in v.covered(r):
if p not in out:
out.append(p)
return sorted(out)
def classify(s, st):
"""R3.5: (sensitive, only an owner may make it)"""
cur = s.settings.get(st.scope())
cur = cur[1] if cur is not None else {}
sensitive = owner = False
for k, val in st.caps:
old = cur.get(k)
if val == ("on", True):
if old != ("on", True):
sensitive = True
owner = owner or k in ("ssh.host", "users", "sudo")
elif val == ("on", False):
pass
elif val[0] == "days":
was = old[1] if old is not None and old[0] == "days" else 400
if val[1] < was:
sensitive = owner = True
else:
if old != val:
sensitive = owner = True
return sensitive, owner
def settings_version_ok(s, st):
if st.machine is not None:
# (WP7) one machine's settings: an active machine, named with its current place
d = s.now().device(st.machine)
need(d is not None and d.kind == MACHINE and d.status == ACTIVE, "R6.14",
lambda: "%s is not an active machine" % dev_name(st.machine))
need(d.reach == st.place, "R6.14", lambda: "%s is not in %s" % (dev_name(st.machine), place_name(st.place)))
cur = s.settings.get(st.scope())
cur = cur[0] if cur is not None else 0
need(st.version == cur + 1, "R6.12",
lambda: "settings version %d where %d is next (versions only go forward)" % (st.version, cur + 1))
def signer_device(v, m):
"""The signer's view (decide refuses earlier any signer that is not a device of the log, R9.5)."""
d = v.device(m)
if d is None:
raise Refusal("§3.2", "%s is not a device of the log: it signs only its own recovery completion (R9.5)"
% dev_name(m))
return d
def decide(s, seq, e, auth):
"""The plan of an authenticated entry, or Refusal (moxivue-log apply.rs decide)."""
v = s.now()
if seq == 0:
return genesis_rule(s, e)
need(not s.deleted, "decision 13", "the account is deleted")
need(e.id not in s.ids, "R7.4", "an entry id seen before: a replay")
need(not auth.stale_key, "R2.7", "signed with a log key the device has since rotated")
need(auth.proofs_ok, "R2.2", "a proof of possession does not verify")
now = entry_time(v, e)
base = e.base
me = e.signer_id()
if me is not None:
if s.device(me) is not None:
may(me, ("sign", e.kind == "leave"), None, v)
else:
# a pending recovery request's console signs nothing but its own completion (R9.5)
need(e.kind == "recovery-complete", "§3.2", lambda: "%s is not a device of the log: it signs only its own"
" recovery completion (R9.5)" % dev_name(me))
k = e.kind
b = e.b
if k == "genesis":
raise Refusal("R8.1", "only the first entry is a genesis")
if k == "propose":
return propose_entry(s, v, seq, me, b["change"], base, e.hash)
if k == "confirm":
p = s.proposals.get(b["proposal"])
if p is None:
raise Refusal("R3.9", "no such proposal")
need(v.pending(p) and now <= p.deadline(), "R3.9", "the proposal is not pending (finished, lapsed or void)")
need(not any(c[1] == me for c in p.confirmations), "R3.1", "confirmed by this device already")
need(base.size > p.seq, "R3.9", "the confirmation's base does not hold the proposal")
held(v, p)
may(me, ("confirm",), ("proposal", p), v)
try:
may(me, ("confirm",), ("proposal", p), s.at(base.size))
except Refusal as r:
raise Refusal(r.rule, "at its base: " + r.why)
return ("confirm", p.hash, p.confirmed is None)
if k == "refuse":
p = s.proposals.get(b["proposal"])
if p is None:
raise Refusal("R3.13", "no such proposal")
need(v.pending(p), "R3.13", "the proposal is not pending")
d = signer_device(v, me)
need(dv_owner_admin(d), "R3.13", "only owner/admin devices refuse")
counts = p.change.kind == "join" and is_ok(may, me, ("refuse",), ("proposal", p), v)
return ("refuse", p.hash, counts)
if k == "withdraw":
p = s.proposals.get(b["proposal"])
if p is None:
raise Refusal("R3.9", "no such proposal")
need(me == p.proposer, "R3.9", "only the proposer withdraws")
need(v.pending(p), "R3.9", "the proposal is not pending")
return ("withdraw", p.hash)
if k == "finish":
return finish(s, v, seq, me, b["proposal"], b["gens"], base, now)
if k == "machine-join":
may(me, ("manage",), ("place", b["place"]), v)
need(not v.setup(), "R3.12", "no machine joins while the account is in set-up (D-5)")
mach = b["machine"]
fresh_keys(s, mach.all_keys())
need(mach.account == s.account, "R2.2", "the machine's key statement names another account")
return ("new-device", new_machine(s, seq, b["place"], mach), None)
if k == "token-issue":
may(me, ("manage",), ("place", b["place"]), v)
need(not v.setup(), "R8.10", "no install tokens are issued in set-up (R3.12)")
need(len(b["tokens"]) > 0, "R8.10", "at least one token")
need(b["expiry"] > base_later(base) and b["expiry"] <= base_later(base) + TOKEN_MAX, "R8.10",
"a token expires within 24 hours of its issue")
fresh_keys(s, b["tokens"])
return ("token-issue",)
if k == "token-revoke":
idx = []
for key in b["tokens"]:
i = next((j for j, t in enumerate(s.tokens) if t.issue == b["issue"] and t.key == key), None)
if i is None:
raise Refusal("R8.10", "no such token in that issue entry")
may(me, ("manage",), ("place", s.tokens[i].place), v)
need(not s.tokens[i].used and not s.tokens[i].revoked, "R8.10", "token already used or revoked")
need(i not in idx, "R8.10", "a token named twice")
idx.append(i)
need(len(idx) > 0, "R8.10", "no token named")
return ("token-revoke", idx)
if k == "token-join":
i = next((j for j, t in enumerate(s.tokens) if t.issue == b["issue"] and t.key == b["token"]), None)
if i is None:
raise Refusal("R8.10", "no such token")
t = s.tokens[i]
need(t.issue < base.size, "R8.10", "the issue entry is not in the checkpoint the join names (not cosigned)")
need(not t.revoked, "R8.10", "the token was revoked")
need(not t.used, "R8.10", "the token was used already: refused as reused (alarm, FA-2)")
need(now <= t.expiry, "R8.10", "the token has expired")
need(not v.setup(), "R3.12", "no machine joins while the account is in set-up")
need(v.place_exists(t.place), "R8.10", "the token's place does not exist")
mach = b["machine"]
need(mach.account == s.account, "R8.10", "the machine's statement names another account")
fresh_keys(s, mach.all_keys())
return ("new-device", new_machine(s, seq, t.place, mach), i)
if k == "leave":
d = signer_device(v, me)
a = After(v)
a.changed[me] = d._replace(status=REMOVED)
need(a.floor_ok(), "R3.12", "this leaving would leave fewer than two whole-account owner/admin devices, one"
" an owner")
return ("leave",)
if k == "place-new":
may(me, ("manage",), ("place", b["parent"]), v)
depth = place_depth(s, b["parent"])
need((depth if depth is not None else 9) < 2, "§3.1", "places are the account, sites in it and groups in a site")
pid = seq
g = b["gen"]
need(g.place == pid, "R5.1", "generation 0 of the new place")
a = After(v)
a.new_place = (pid, b["parent"])
author = is_ok(may, me, ("generation",), ("place", b["parent"]), v)
check_gen(v, g, "new-place", a.recipients(pid), author)
return ("place-new",)
if k == "place-name":
may(me, ("manage",), ("place", b["place"]), v)
return ("rename",)
if k == "person-name":
may(me, ("manage",), ("place", ACCOUNT), v)
need(b["person"] in s.persons, "§3.1", "no such person")
return ("rename",)
if k == "generation":
g = b["gen"]
need(g.reason in ("rotation", "freeze-end", "scheduled"), "R5.7",
lambda: "a %s generation is carried by its own entry" % g.reason)
need(not waits_for_removal(v, g.place, None), "R3.10",
"a confirmed removal is pending for a console of this place: other generations wait (W-22)")
need(g.reason != "scheduled" or not carries_pending(v, g.place), "R5.7",
"no scheduled generation while a confirmed change carrying one for this place is pending")
author = is_ok(may, me, ("generation",), ("place", g.place), v)
check_gen(v, g, g.reason, v.recipients(g.place), author)
return ("generation",)
if k == "rotate":
d = signer_device(v, me)
keys = b["keys"]
need((d.kind == MACHINE) == (keys.k4 is not None) and (keys.k4 is not None) == (keys.k14 is not None), "R2.7",
"a machine rotates K1-K4 and K14, a console K1-K3")
need(keys.account == s.account, "R2.7", "the new keys' statement names another account")
fresh_keys(s, keys.all_keys())
if dv_console(d) and dv_owner_admin(d):
want = [p for p in v.covered(d.reach) if not waits_for_removal(v, p, None)]
else:
want = []
need([g.place for g in b["gens"]] == want, "R2.7",
"an owner/admin console's rotation carries a new generation for every place it reaches (F11)")
for g in b["gens"]:
check_gen(v, g, "rotation", v.recipients(g.place), True)
return ("rotate",)
if k == "support":
return ("nothing",)
if k == "settings":
st = b["st"]
may(me, ("manage",), ("place", st.place), v)
settings_version_ok(s, st)
need(not classify(s, st)[0], "R3.5", "this settings change is sensitive: it needs a proposal and a confirmation")
return ("settings",)
if k == "ssh-grant":
may(me, ("sshgrant",), ("device", b["machine"]), v)
mv = v.device(b["machine"])
need(ssh_host_on(s, mv.reach), "R6.14", "ssh.host is not on for this machine's place")
h = v.acting(b["holder"])
if h is None:
raise Refusal("R6.14", "the holder is not an active console")
need(dv_console(h) and v.covers(h.reach, mv.reach), "R6.14", "the holder's reach does not cover the machine")
hr = h.role
need(hr in (OWNER, ADMIN, OPERATOR), "R6.14", lambda: "a %s opens no SSH session" % hr)
need(b["expiry"] > now and b["expiry"] <= base_later(base) + SSH_GRANT_MAX, "R6.14",
"a grant lasts at most 30 days")
need(not b["unattended"] or hr == OPERATOR, "R6.14",
"unattended keys only on automation devices of role operator or lower")
return ("ssh-grant",)
if k == "recovery-request":
need(not auth.stale_paper, "R9.1", "made with a recovery paper that has been replaced")
r = s.recovery
need(not r.suspended, "R9.4", "the recovery paper is suspended after a cancelled request until it is replaced")
for n in b["named"]:
d = v.device(n)
if d is None:
raise Refusal("R9.2", "%s is not in the log" % dev_name(n))
need(dv_console(d) and d.status == ACTIVE and d.person == r.holder, "R9.2",
lambda: "%s is not an active console of the holder's person (D-8)" % dev_name(n))
need(b["snapshot"] == snapshot_at(v), "R9.2", "the snapshot differs from the owner/admin devices active here")
need(b["console"].account == s.account, "R9.2", "the console's statement names another account")
fresh_keys(s, b["console"].all_keys())
other = next((x.hash for _, x in sorted(s.requests.items()) if x.status == R_PENDING and x.key == r.key), None)
return ("request", other)
if k == "cancel":
r = s.requests.get(b["request"])
if r is None:
raise Refusal("R9.4", "no such request")
need(r.status == R_PENDING, "R9.4", "the request is not pending")
may(me, ("cancel",), ("request", r), v)
end, named = cancel_effect(r, [me], r.named_cancels)
return ("cancel", r.hash, end, named)
if k == "recovery-complete":
return complete(s, v, seq, me, b, auth)
raise Refusal("R2.6", "unknown entry kind")
def cancel_effect(r, new, before_named):
snap = {x[0]: x for x in r.snapshot}
def fact(d):
x = snap.get(d)
return None if x is None else (d, d in r.named, x[2] == ACCOUNT)
facts = [f for f in (fact(d) for d in list(before_named) + list(new)) if f is not None]
named_any = any(f[1] for f in (fact(d) for d in new) if f is not None)
return cancels_end(facts), named_any
def snapshot_at(v):
return [(i, sha256(v.s.devices[i].k1), d.reach) for i, d in v.devices()
if dv_console(d) and dv_owner_admin(d) and d.status == ACTIVE]
def ssh_host_on(s, place):
cur = place
for _ in range(4):
if cur is None:
break
c = s.settings.get(cur)
if c is not None and "ssh.host" in c[1]:
return c[1]["ssh.host"] == ("on", True)
pl = s.places.get(cur)
cur = pl.parent if pl is not None else None
return False
def new_machine(s, seq, place, k):
view = DV(MACHINE, None, place, None, False, ACTIVE)
return Device(seq, view, k.k1, k.k2, k.k3, k.k4, k.k14, s.recovery_epoch(), seq + 1)
def genesis_rule(s, e):
b = e.b
need(s.len() == 0, "R8.1", "a genesis only as the first entry")
fresh_keys(s, b["console"].all_keys() + [b["recovery"]])
need(b["console"].account == bytes(32), "R8.1", "the genesis console's statement names no account (it makes it)")
g = b["gen"]
need(g.place == ACCOUNT and g.n == 0 and g.reason == "new-place", "R8.1", "generation 0 of the whole account")
need(g.recipient_ids() == [0], "R5.6", "the first console is the only recipient")
return ("genesis",)
def held(v, p):
"""R3.11: a removal of a device that is the target of an earlier confirmed removal waits for that one."""
if p.change.kind == "remove":
t = p.change.f["target"]
other = any(q.hash != p.hash and v.confirmed_pending(q) for q in v.s.removals_of.get(t, ()))
need(not other, "R3.11", lambda: "%s is the target of an earlier confirmed removal: this one is held" % dev_name(t))
def authority(s, v, m, c):
md = v.device(m)
if md is None:
raise Refusal("§3.2", "not in the log")
t = c.target()
if t is not None:
need(t != m, "R3.2", "a device is never the target of its own proposal (its leaving is its own entry)")
k = c.kind
f = c.f
if k == "join":
return may(m, ("propose", "join", (f["t"].role, f["t"].reach)), ("place", f["t"].reach), v)
if k == "remove":
return may(m, ("propose", "remove"), ("device", f["target"]), v)
if k == "role":
return may(m, ("propose", "role", f["role"]), ("device", f["target"]), v)
if k == "reach":
return may(m, ("propose", "reach", f["reach"]), ("device", f["target"]), v)
if k == "move":
return may(m, ("propose", "move", f["place"]), ("device", f["target"]), v)
if k == "person":
return may(m, ("propose", "person"), ("device", f["target"]), v)
if k == "history":
may(m, ("propose", "history"), None, v)
need(md.person != f["person"], "R6.13", "never signed by a device of the grantee's person")
return
if k == "recovery-replace":
return may(m, ("propose", "replace-recovery"), None, v)
if k == "settings":
what = "settings-owner" if classify(s, f["st"])[1] else "settings"
return may(m, ("propose", what), ("place", f["st"].place), v)
if k == "delete-records":
return may(m, ("propose", "delete-records"), ("place", f["place"]), v)
if k in ("switch-over", "escalate"):
return may(m, ("propose", "switch-over"), None, v)
if k == "delete-account":
return may(m, ("propose", "delete-account"), None, v)
raise Refusal("R2.6", "unknown change")
def propose_entry(s, v, seq, m, c, base, h):
md = signer_device(v, m)
authority(s, v, m, c)
waiting = person = None
f = c.f
if c.kind == "join":
t = f["t"]
need(t.account == s.account, "R8.2", "the transcript names another account")
need(t.inviter == s.devices[m].k1, "R8.2", "the transcript's inviting key is not the proposer's (F1)")
fresh_keys(s, t.joiner.all_keys())
if t.person == "new":
person = seq
pid = seq
else:
pid = md.person
view = DV(CONSOLE, t.role, t.reach, pid, t.automation, WAITING)
waiting = Device(seq, view, t.joiner.k1, t.joiner.k2, t.joiner.k3, None, None, s.recovery_epoch(), seq + 1)
elif c.kind == "person":
need(f["person"] in s.persons, "R3.5", "no such person")
d = v.device(f["target"])
need((d.person if d is not None else None) != f["person"], "R3.5", "it is that person's already")
elif c.kind == "history":
need(f["person"] in s.persons, "R6.13", "no such person")
need(len(f["consoles"]) > 0, "R6.13", "the grant names consoles")
for d in f["consoles"]:
x = v.device(d)
need(x is not None and dv_console(x) and x.status == ACTIVE and x.person == f["person"], "R6.13",
lambda: "%s is not an active console of the grantee" % dev_name(d))
elif c.kind == "recovery-replace":
need(f["holder"] in s.persons, "R9.1", "no such holder person")
fresh_keys(s, [f["key"]])
elif c.kind == "settings":
need(classify(s, f["st"])[0], "R3.5", "not a sensitive settings change: a settings entry, no proposal")
settings_version_ok(s, f["st"])
elif c.kind == "switch-over":
raise Refusal("R10b.3", "suite %d is not a newer suite this build knows (5.0 has suite 1 only)" % f["suite"])
elif c.kind == "escalate":
raise Refusal("R10b.9", "no normal switch-over is pending to escalate")
t = c.target()
td = v.device(t) if t is not None else None
p = Proposal()
p.hash, p.seq, p.proposer, p.change, p.base = h, seq, m, c, base
p.status, p.confirmations, p.confirmed, p.ended = PENDING, [], None, None
p.freezes = freeze_target(c, td.role if td is not None else None)
return ("propose", p, waiting, person)
def privileged(v, p):
"""R3.6: None when the change is not privileged, else (person known?, person)."""
c = p.change
if c.kind == "join":
if c.f["t"].person == "new":
return (False, None)
d = v.device(p.proposer)
return (d is not None and d.person is not None, d.person if d is not None else None)
if c.kind == "role":
d = v.device(c.f["target"])
if d is None:
return None
r = c.f["role"]
promotion = owner_admin(r) and (d.role is None or _rank_cmp(d.role) < _rank_cmp(r))
return (d.person is not None, d.person) if promotion else None
if c.kind == "reach":
d = v.device(c.f["target"])
if d is None:
return None
if v.covers(c.f["reach"], d.reach) and c.f["reach"] != d.reach:
return (d.person is not None, d.person)
return None
return None
# Finding GEN-1 (moxivue-log apply.rs GEN1): how R5.1's "never the target of the change it carries" reads for a
# change that gives its target more access. In force: option A, "reducing-only" (Mo, 2026-10-20): the target of a
# role, reach or place change is left out of the generations its own change carries only when the change reduces its
# access (a demotion; a reach that no longer covers all it covered). "literal" was the chapter's earlier text.
GEN1 = "reducing-only"
def gen_excludes_target(c, before, v):
if c.kind == "role":
if GEN1 == "literal":
return True
return before.role is None or is_demotion(before.role, c.f["role"])
if c.kind in ("reach", "move"):
if GEN1 == "literal":
return True
return not v.covers(c.f["reach"] if c.kind == "reach" else c.f["place"], before.reach)
return False
def effect_of(v, p):
changed = []
exclude = None
c = p.change
def dev(i):
d = v.device(i)
if d is None:
raise Refusal("§3.1", "%s is not in the log" % dev_name(i))
return d
if c.kind == "join":
d = dev(p.seq)
need(d.status == WAITING, "R3.13", "the joiner is no longer waiting")
changed.append((p.seq, d._replace(status=ACTIVE)))
elif c.kind == "remove":
d = dev(c.f["target"])
need(d.status == ACTIVE, "§3.1", "the target is not active")
changed.append((c.f["target"], d._replace(status=REMOVED)))
elif c.kind == "role":
d = dev(c.f["target"])
changed.append((c.f["target"], d._replace(role=c.f["role"])))
exclude = c.f["target"] if gen_excludes_target(c, d, v) else None
elif c.kind in ("reach", "move"):
d = dev(c.f["target"])
changed.append((c.f["target"], d._replace(reach=c.f["reach"] if c.kind == "reach" else c.f["place"])))
exclude = c.f["target"] if gen_excludes_target(c, d, v) else None
elif c.kind == "person":
d = dev(c.f["target"])
changed.append((c.f["target"], d._replace(person=c.f["person"])))
return changed, exclude
def finish(s, v, seq, m, h, gens, base, now):
p = s.proposals.get(h)
if p is None:
raise Refusal("R3.8", "no such proposal")
need(m == p.proposer, "R3.8", "only the proposer finishes its proposal")
need(v.pending(p), "R3.9", "the proposal is not pending (finished, withdrawn, void)")
need(now <= p.deadline() and base_later(base) <= p.deadline(), "R3.9", "the proposal has lapsed")
need(base.size > p.seq, "R3.10", "the finishing entry's base does not hold the proposal")
places = [g.place for g in gens]
target = p.change.target()
if target is None and p.change.kind == "join":
target = p.seq
for k in range(base.size, seq):
t = s.touches[k]
need(not conflicts(t.proposal, t.devices, t.gens, h, target, places), "R3.10",
lambda: "entry %d after the base conflicts with this change (V1)" % k)
held(v, p)
try:
authority(s, v, m, p.change)
except Refusal as r:
raise Refusal(r.rule, "the proposer may no longer make this change: " + r.why)
counting = [c for c in p.confirmations
if is_ok(may, c[1], ("confirm",), ("proposal", p), v)
and is_ok(may, c[1], ("confirm",), ("proposal", p), s.at(c[2].size))]
exception = (p.change.kind == "join" and owner_admin(p.change.f["t"].role) and s.single_exception
and v.owner_admin_count() == 1)
if not exception:
need(len(counting) > 0, "R3.1", "no counting confirmation from an eligible confirmer")
pr = privileged(v, p)
if pr is not None and pr[0]:
pp = pr[1]
place = change_place(p, v)
exists_other = any(dv_console(d) and dv_owner_admin(d) and not d.automation and d.status == ACTIVE
and d.person != pp and v.covers(d.reach, place) for _, d in v.devices())
if exists_other:
def other_person(c):
d = v.device(c[1])
return (d.person if d is not None else None) != pp
need(any(other_person(c) for c in counting), "R3.6",
"a privileged join or promotion needs a confirmer of another person where one exists (I1, V2)")
if p.change.kind == "join":
need(not v.setup() or owner_admin(p.change.f["t"].role), "R3.12", "in set-up only owner and admin consoles join")
if p.change.kind == "settings":
settings_version_ok(s, p.change.f["st"])
changed, exclude_from_gens = effect_of(v, p)
a = After(v)
a.unfrozen = p.freezes
for i, d in changed:
a.changed[i] = d
need(a.floor_ok(), "R3.12", "this change would leave fewer than two whole-account owner/admin devices, one an "
"owner (D1)")
want = gen_places(v, p.change)
need(places == want, "R5.3", lambda: "the change carries generations for %r (every place its target reaches)" % want)
adevs = a.devices()
for g in gens:
need(not waits_for_removal(v, g.place, p), "R3.10",
lambda: "a confirmed removal is pending for a console of %s: this change waits (W-22)" % place_name(g.place))
rec = a.recipients(g.place)
if exclude_from_gens is not None:
rec = [d for d in rec if d != exclude_from_gens]
author = any(i == m and dv_console(d) and dv_owner_admin(d) and d.status == ACTIVE and a.covers(d.reach, g.place)
for i, d in adevs)
check_gen(v, g, "membership", rec, author)
return ("finish", h, changed, gens)
def complete(s, v, seq, m, b, auth):
h = b["request"]
r = s.requests.get(h)
if r is None:
raise Refusal("R9.5", "no such request")
need(m == r.seq, "R9.5", "only the requesting console completes")
need(r.status == R_PENDING, "R9.5", "the request is not pending (ended, void or completed)")
start = b["start"]
need(s.known_base(start) and start.size > r.seq and start.size <= seq, "R9.3",
"the delay starts at a verified checkpoint that holds the request")
lists = b["lists"]
need(len(lists) == 2 and lists[0].witness == 0 and lists[1].witness == 1, "R9.5", "both witnesses' cancel lists")
need(all(auth.lists_ok) and len(auth.lists_ok) == 2, "R9.5", "a cancel list not signed by its witness")
witnessed = [d for l in auth.cancels for _, d in l]
end, named_any = cancel_effect(r, witnessed, r.named_cancels)
need(not end, "R9.5", "a cancel list holds a cancel that ends the request")
snap = {x[0]: x for x in r.snapshot}
extended = (r.extended or named_any
or any(d in snap and snap[d][2] != ACCOUNT and d not in r.named for d in witnessed))
delay = DELAY_D + (DELAY_D2 if extended else 0)
due = base_later(start) + delay
need(all(l.time >= due for l in lists), "R9.5",
lambda: "the cancel lists must be dated after the delay ended (%d days from the request's checkpoint)"
% (delay // DAY))
fresh_keys(s, [b["key"]])
a = After(v)
a.changed[m] = DV(CONSOLE, OWNER, ACCOUNT, r.holder, False, ACTIVE)
for n in r.named:
d = v.device(n)
if d is not None:
a.changed[n] = d._replace(status=REMOVED)
need([g.place for g in b["gens"]] == v.places(), "R9.6", "new generations for every place")
for g in b["gens"]:
check_gen(v, g, "recovery", a.recipients(g.place), True)
return ("complete", h)
def set_gen(s, place, n, h, seq):
s.gens[place] = (n, h, seq)
s.gen_list.setdefault(place, []).append((n, seq))
def void_proposals_of(s, d, size):
voided = []
for p in s.proposals.values():
if p.proposer == d and p.status == PENDING and p.ended is None:
p.status = VOID
p.ended = size
if p.change.kind == "join":
voided.append(p.seq)
for w in voided:
x = s.devices.get(w)
if x is not None and x.view.status == WAITING:
s.set_view(w, size, x.view._replace(status=REMOVED))
def apply_settings(s, st):
e = s.settings.setdefault(st.scope(), [0, {}])
e[0] = st.version
for k, val in st.caps:
e[1][k] = val
def commit(s, seq, e, plan):
"""Applies a decided entry (moxivue-log apply.rs commit). Never refuses."""
size = seq + 1
me = e.signer_id()
touch = Touch()
seen = []
def mark(keys):
for k in keys:
for half in key_halves(k):
seen.append(sha256(half))
kind = plan[0]
b = e.b
if kind == "genesis":
con = b["console"]
mark(con.all_keys())
mark([b["recovery"]])
s.account = e.hash
s.name = b["name"]
s.persons[0] = b["person"]
s.places[ACCOUNT] = Place(ACCOUNT, None, b["name"], size)
s.add_device(Device(0, DV(CONSOLE, OWNER, ACCOUNT, 0, False, ACTIVE), con.k1, con.k2, con.k3, None, None, 0,
size))
s.recovery = Recovery(b["recovery"], 0, False, 0)
set_gen(s, ACCOUNT, 0, b["gen"].hash, seq)
s.single_exception = True
touch.devices.append(0)
touch.gens.append(ACCOUNT)
elif kind == "propose":
_, p, waiting, person = plan
if person is not None:
s.persons[person] = "person %d" % person
if waiting is not None:
mark([waiting.k1, waiting.k2, waiting.k3])
touch.devices.append(waiting.id)
s.add_device(waiting)
if p.change.kind == "recovery-replace":
mark([p.change.f["key"]])
s.add_proposal(p)
elif kind == "confirm":
_, h, first = plan
p = s.proposals[h]
p.confirmations.append((seq, me, e.base))
if first:
p.confirmed = size
if p.freezes is not None:
void_proposals_of(s, p.freezes, size)
elif kind == "refuse":
_, h, counts = plan
if counts:
p = s.proposals[h]
p.status = REFUSED_P
p.ended = size
w = p.seq
s.set_view(w, size, s.devices[w].view._replace(status=REMOVED))
touch.proposal = h
touch.devices.append(w)
elif kind == "withdraw":
h = plan[1]
p = s.proposals[h]
p.status = WITHDRAWN
p.ended = size
touch.proposal = h
if p.change.kind == "join":
w = p.seq
s.set_view(w, size, s.devices[w].view._replace(status=REMOVED))
touch.devices.append(w)
elif kind == "finish":
_, h, changed, gens = plan
for i, view in changed:
s.set_view(i, size, view)
touch.devices.append(i)
if view.status == REMOVED:
void_proposals_of(s, i, size)
for g in gens:
set_gen(s, g.place, g.n, g.hash, seq)
touch.gens.append(g.place)
p = s.proposals[h]
p.status = FINISHED
p.ended = size
touch.proposal = h
c = p.change
if c.kind == "recovery-replace":
r = s.recovery
old = r.key
r.key = c.f["key"]
r.holder = c.f["holder"]
r.suspended = False
for q in s.requests.values():
if q.status == R_PENDING and q.key == old:
q.status = R_VOID
elif c.kind == "settings":
apply_settings(s, c.f["st"])
elif c.kind == "delete-account":
s.deleted = True
elif kind == "new-device":
_, d, token = plan
mark([d.k1, d.k2, d.k3])
mark([d.k4 if d.k4 is not None else b"", d.k14 if d.k14 is not None else b""])
touch.devices.append(d.id)
s.add_device(d)
if d.k4 is not None:
s.k4_hist.setdefault(d.id, []).append((size, d.k4))
if token is not None:
s.tokens[token].used = True
elif kind == "token-issue":
for key in b["tokens"]:
mark([key])
t = Token()
t.key, t.issue, t.place, t.expiry, t.revoked, t.used = key, seq, b["place"], b["expiry"], False, False
s.tokens.append(t)
elif kind == "token-revoke":
for i in plan[1]:
s.tokens[i].revoked = True
elif kind == "leave":
s.set_view(me, size, s.devices[me].view._replace(status=REMOVED))
void_proposals_of(s, me, size)
touch.devices.append(me)
elif kind == "place-new":
pid = seq
s.places[pid] = Place(pid, b["parent"], b["name"], size)
set_gen(s, pid, 0, b["gen"].hash, seq)
touch.gens.append(pid)
elif kind == "rename":
if e.kind == "place-name":
pl = s.places.get(b["place"])
if pl is not None:
pl.name = b["name"]
else:
s.persons[b["person"]] = b["name"]
elif kind == "generation":
g = b["gen"]
set_gen(s, g.place, g.n, g.hash, seq)
touch.gens.append(g.place)
elif kind == "rotate":
keys = b["keys"]
mark(keys.all_keys())
d = s.devices[me]
d.old_k1.append(d.k1)
d.k1, d.k2, d.k3, d.k4, d.k14 = keys.k1, keys.k2, keys.k3, keys.k4, keys.k14
if keys.k4 is not None:
s.k4_hist.setdefault(me, []).append((size, keys.k4))
touch.devices.append(me)
for g in b["gens"]:
set_gen(s, g.place, g.n, g.hash, seq)
touch.gens.append(g.place)
elif kind == "nothing":
pass
elif kind == "settings":
apply_settings(s, b["st"])
elif kind == "ssh-grant":
if b["unattended"]:
hd = s.devices[b["holder"]]
s.set_view(b["holder"], size, hd.view._replace(automation=True))
touch.devices.append(b["holder"])
elif kind == "request":
mark(b["console"].all_keys())
r = s.recovery
q = Request()
q.hash, q.seq, q.console, q.named, q.snapshot = e.hash, seq, b["console"], b["named"], b["snapshot"]
q.holder, q.key, q.status, q.extended, q.named_cancels = r.holder, r.key, R_PENDING, False, []
other = plan[1]
if other is not None:
q.status = R_VOID
x = s.requests.get(other)
if x is not None:
x.status = R_VOID
s.requests[e.hash] = q
elif kind == "cancel":
_, h, end, named = plan
r = s.requests[h]
if named and me not in r.named_cancels:
r.named_cancels.append(me)
if end:
r.status = R_ENDED
if s.recovery is not None:
s.recovery.suspended = True
else:
r.extended = True
elif kind == "complete":
h = plan[1]
mark([b["key"]])
r = s.requests[h]
epoch = s.recovery_epoch() + 1
view = DV(CONSOLE, OWNER, ACCOUNT, r.holder, False, ACTIVE)
d = Device(r.seq, view, r.console.k1, r.console.k2, r.console.k3, None, None, epoch, size)
s.add_device(d)
touch.devices.append(r.seq)
for n in r.named:
x = s.devices.get(n)
if x is not None and x.view.status == ACTIVE:
s.set_view(n, size, x.view._replace(status=REMOVED))
void_proposals_of(s, n, size)
touch.devices.append(n)
for g in b["gens"]:
set_gen(s, g.place, g.n, g.hash, seq)
touch.gens.append(g.place)
old = s.recovery.key if s.recovery is not None else None
s.recovery = Recovery(b["key"], r.holder, False, epoch)
for q in s.requests.values():
if q.hash == h:
q.status = R_COMPLETED
elif q.status == R_PENDING and q.key == old:
q.status = R_VOID
else:
raise AssertionError("plan " + kind)
s.keys_seen.update(seen)
s.touches.append(touch)
n = s.at(size).owner_admin_count()
if e.kind == "recovery-complete":
s.single_exception = n == 1
elif n != 1:
s.single_exception = False
if s.floor_since is None:
d, o = s.at(size).shape()
if d >= FLOOR_DEVICES and o >= FLOOR_OWNERS:
s.floor_since = size
def append(s, leaf_bytes, w):
"""Appends one leaf: authenticates, decides and applies it. Returns None (accepted) or the Refusal; raises Fail."""
if s.len() >= MAX_ENTRIES:
raise Fail("log longer than the verifier takes")
try:
seq, prev, signed = leaf_from(leaf_bytes)
if seq != s.len():
raise Fail("leaf at position %d where %d was next" % (seq, s.len()))
if prev != (s.tree.leaves[-1] if s.tree.leaves else bytes(32)):
raise Fail("leaf does not chain to the previous one")
sg = signed_from(signed)
except DecodeError as e:
raise Fail("not a well-formed log structure: %s" % e.args[0])
auth = authenticate(s, seq, sg, w)
e = sg.entry
try:
plan = decide(s, seq, e, auth)
commit(s, seq, e, plan)
out = None
except Refusal as r:
s.refused.append((seq, r))
s.touches.append(Touch())
out = r
if e.base is not None:
cur = s.now().time()
t = max(base_later(e.base), cur)
if t > cur:
s.now_sizes.append(seq + 1)
s.now_times.append(t)
s.ids.add(e.id)
s.tree.push(leaf_hash(leaf_bytes))
return out
class Report:
def __init__(self, state, failure, refused, cosigned, entries):
self.state = state
self.failure = failure # (position, reason) or None
self.refused = refused # [(position, Refusal)]
self.cosigned = cosigned
self.entries = entries
def clean(self):
return self.failure is None and not self.refused
def verify(leaves, checkpoints, w):
"""Verifies a whole log from its genesis (moxivue-log verify.rs verify)."""
s = State()
failure = None
cps = sorted(checkpoints, key=lambda c: c[0].size)
nxt = 0
def add(upto):
nonlocal nxt
while nxt < len(cps) and cps[nxt][0].size <= upto:
add_checkpoint(s, cps[nxt][0], cps[nxt][1], w)
nxt += 1
for i, leaf in enumerate(leaves):
try:
add(i)
append(s, leaf, w)
except Fail as f:
failure = (i, str(f))
break
if failure is None:
n = s.len()
try:
add(n)
if nxt < len(cps):
failure = (n, "checkpoint: checkpoint of size %d beyond the log" % cps[nxt][0].size)
except Fail as f:
failure = (n, str(f))
return Report(s, failure, list(s.refused), s.cosigned(), len(leaves))
# ======================================================================== records and status (moxivue-data record.rs,
# console.rs R6.8): checked against the verified log the way a console checks them before opening. This checker
# holds no keys, so it does not open them: what a sealed record or status says (its readings, a status's newest
# record sequence) is the consoles' to read.
L_RECORD, L_STATUS = "moxivue/record/v1", "moxivue/status/v1"
class Item:
__slots__ = ("kind", "account", "machine", "place", "n", "i", "seq", "header", "sealed", "sig")
def item_from(b):
"""A record or status item (strict: exact keys, the suite's tags, a period 0..7); DecodeError otherwise."""
try:
v = cdec(b)
except CanonError as e:
_bad("encoding: " + e.args[0])
kind = "status" if type(v) is dict and v.get("t") == "status" else "record"
if type(v) is not dict or len(v) != 4 or v.get("t") != kind:
_bad("not a " + kind)
h = v.get("h")
if type(h) is not dict or len(h) != 9 or h.get("t") != kind + "-header":
_bad("not a %s header" % kind)
it = Item()
it.kind = kind
it.account = _hash_of(h.get("acct"))
if it.account is None:
_bad("account")
for k in ("machine", "place", "n", "i", "seq", "seal", "sig"):
if not _is_u64(h.get(k)):
_bad(k)
if h["i"] >= 8:
_bad("period outside 0..=7")
if h["seal"] != int.from_bytes(TAG_SEAL, "big") or h["sig"] != int.from_bytes(TAG_SIG, "big"):
_bad("algorithm tags of another suite")
it.machine, it.place, it.n, it.i, it.seq = h["machine"], h["place"], h["n"], h["i"], h["seq"]
it.header = h
it.sealed, it.sig = v.get("ct"), v.get("sig")
if type(it.sealed) is not bytes or type(it.sig) is not bytes:
_bad("ct or sig")
if sealed_ok(it.sealed):
_bad("sealed part refused")
if _tag(it.sig, TAG_SIG) or len(it.sig) != SIG_LEN or mldsa_sig_decode(it.sig[2 + ED_SIG_LEN:]) is None:
_bad("signature refused")
return it
def item_verifies(it, k4):
signed = cenc({"t": it.kind + "-signed", "h": it.header, "ct": sha256(it.sealed)})
return hybrid_verify(k4, L_RECORD if it.kind == "record" else L_STATUS, signed, it.sig)
def _within(hist, frm, to):
out = []
for i, (start, key) in enumerate(hist):
end = hist[i + 1][0] if i + 1 < len(hist) else None
if (to is None or start < to) and (end is None or end > frm):
out.append(key)
return out
def records_export_from(b):
try:
v = cdec(b)
except CanonError as e:
_bad("records export: " + e.args[0])
if type(v) is not dict or len(v) != 3 or v.get("t") != "records-export" or v.get("v") != 1 \
or type(v.get("items")) is not list or not all(type(x) is bytes for x in v["items"]):
_bad("not a records export")
return v["items"]
# ---- STORED-1 (key chapter §3.2): a requester console's own action record (stored.read), K1 under action-record,
# kept by the Dispatcher next to the machine's records (moxivue-data action.rs)
L_ACTION_RECORD = "moxivue/action-record/v1"
ACTION_KINDS = ("stored.read", "recording.play", "recording.export", "recording.reseal", "audit.export")
def _text_ok(t):
return type(t) is str and 0 < len(t.encode()) <= 200 and not any(ord(c) < 32 or 0x7f <= ord(c) < 0xa0 for c in t)
def action_check(s, b):
"""None when the action record `b` holds against the verified state `s` (as ActionRecord::accept), else why."""
try:
v = cdec(b)
except CanonError:
return "format"
if type(v) is not dict or set(v) != {"t", "rec", "sig"} or v["t"] != "action-record-signed" \
or type(v["rec"]) is not bytes or type(v["sig"]) is not bytes:
return "format"
body = v["rec"]
try:
r = cdec(body)
except CanonError:
return "format"
keys = {"t", "account", "requester", "machine", "kind", "what", "from", "to", "path", "time", "base"}
if type(r) is not dict or set(r) != keys or r["t"] != "action-record" or cenc(r) != body:
return "format"
if _hash_of(r["account"]) is None or not all(_is_u64(r[k]) for k in ("requester", "machine", "from", "to", "time")):
return "format"
if r["kind"] not in ACTION_KINDS or not _text_ok(r["what"]) or not _text_ok(r["path"]) or r["from"] > r["to"]:
return "format"
try:
base = base_from(r["base"])
except DecodeError:
return "format"
if r["account"] != s.account:
return "account"
if base not in s.checkpoints.get(base.size, []):
return "checkpoint"
d = s.devices.get(r["requester"])
view = d.at(base.size) if d is not None else None
if view is None or view.kind != CONSOLE or view.status != ACTIVE:
return "requester"
if not any(hybrid_verify(k, L_ACTION_RECORD, body, v["sig"]) for k in [d.k1] + list(d.old_k1)):
return "signature"
return None
RECORD_WHY = {
"checkpoint": "an action record naming a checkpoint that is not one of the log's",
"requester": "an action record whose requester was not an active console of the account then",
"format": "not a well-formed record or status",
"account": "another account's item",
"no-generation": "no such generation in the log",
"writer": "the writer is not a machine of the account",
"signature": "the writer's record key does not verify it",
"place": "the writer was not in that place while the generation was current",
"sequence": "sequence not after the last one: replayed or out of order",
}
class RecordCheck:
"""Records and status of an export against a verified log (R6.8): per item None (accepted) or why refused, and
the record sequences missing under each writer's key."""
def __init__(self, s):
self.s = s
self.last = {} # (machine, key hash) -> [record seq, status seq]
self.actions = 0 # requesters' action records that hold (STORED-1)
self.gaps = {} # (machine, key hash) -> [(from, to)]
def check(self, b):
s = self.s
# STORED-1: a requester's action record kept with the records
try:
t = cdec(b)
except CanonError:
t = None
if type(t) is dict and t.get("t") == "action-record-signed":
why = action_check(s, b)
if why is None:
self.actions += 1
return why
try:
it = item_from(b)
except DecodeError:
return "format"
if it.account != s.account:
return "account"
gl = s.gen_list.get(it.place, [])
k = next((j for j, (n, _) in enumerate(gl) if n == it.n), None)
if k is None:
return "no-generation"
frm = gl[k][1] + 1
to = gl[k + 1][1] + 1 if k + 1 < len(gl) else None
dev = s.devices.get(it.machine)
if dev is None or dev.view.kind != MACHINE:
return "writer"
hist = dev.hist
in_place = any(v.reach == it.place and v.status == ACTIVE and (to is None or start < to)
and (j + 1 >= len(hist) or hist[j + 1][0] > frm) for j, (start, v) in enumerate(hist))
key = next((k4 for k4 in _within(s.k4_hist.get(it.machine, []), frm, to) if item_verifies(it, k4)), None)
if key is None:
return "signature"
if not in_place:
return "place"
slot = self.last.setdefault((it.machine, sha256(key)), [0, 0])
x = 0 if it.kind == "record" else 1
if it.seq <= slot[x]:
return "sequence"
if x == 0 and it.seq > slot[0] + 1:
self.gaps.setdefault((it.machine, sha256(key)), []).append((slot[0] + 1, it.seq - 1))
slot[x] = it.seq
return None
def missing(self):
return [(m, a, b) for (m, _), gs in sorted(self.gaps.items()) for a, b in gs]
def check_records_vectors(path):
"""The records vectors of moxivue-data (tests/vectors/records.canon): an owner console's verdicts."""
with open(path, "rb") as f:
v = cdec(f.read())
_, wk, leaves, cps = read_golden(v["golden"])[:4]
rep = verify(leaves, cps, Witnesses.test(wk))
if not rep.clean():
print("the log of the records vectors does not verify")
return False
rc = RecordCheck(rep.state)
bad = 0
items = records_export_from(v["records"])
for i, (b, want) in enumerate(zip(items, v["verdicts"])):
got = rc.check(b)
if (got is None) != want["ok"] or (got or "") != want["why"]:
bad += 1
print("DIFFERENT item %d: %r, the Rust console %r" % (i, got, want))
if [list(x) for x in rc.missing()] != v["missing"]:
bad += 1
print("DIFFERENT missing: %r, the Rust console %r" % (rc.missing(), v["missing"]))
print("records: %d items, %d different" % (len(items), bad))
return bad == 0 and len(items) == len(v["verdicts"])
# ======================================================================== the witnesses (moxivue-log checkpoint.rs, WP4)
# - checkpoint: the C2SP tlog-checkpoint note body "moxivue.com/account/<hex fingerprint>\n<size>\n<base64 root>\n";
# - cosignature: a hybrid signature whose two halves both sign the C2SP cosignature/v1 message
# "cosignature/v1\ntime <t>\n<body>" (Ed25519 strict, ML-DSA-65 pure with an empty context), stored as
# canon({"t": "cosignature", "time": t, "sig": signature}); a checkpoint's proof is canon([witness 1's, witness 2's])
# and each time must be the stamp the base names;
# - cancel list: the witness's hybrid signature under moxivue/cancel-list/v1 over
# canon({"t": "cancel-list-for", "acct": account, "list": the list}).
# The two witnesses built into the apps (crates/moxivue-witness/keys/witnesses.txt): WITNESS_KEYS below holds the
# same once the rollout fills that file in; until then no checkpoint verifies (fail closed).
# (witness 1, witness 2): each {"name": .., "keys": [current, next...], "replaced": None or {"keys": [...],
# "upto": {account bytes: (size, root)}}} (R4.13)
WITNESS_KEYS = None
C2SP_COSIGNATURE_V1 = b"cosignature/v1\ntime "
def checkpoint_body(account, size, root):
return ("moxivue.com/account/%s\n%d\n%s\n" % (account.hex(), size, base64.b64encode(root).decode())).encode()
def cosigned_message(account, size, root, time):
return b"cosignature/v1\ntime %d\n" % time + checkpoint_body(account, size, root)
def is_c2sp_cosignature(msg):
if not msg.startswith(C2SP_COSIGNATURE_V1):
return False
rest = msg[len(C2SP_COSIGNATURE_V1):]
nl = rest.find(b"\n")
if nl < 0:
return False
digits, body = rest[:nl], rest[nl + 1:]
return (0 < len(digits) <= 20 and digits.isdigit() and (len(digits) == 1 or digits[:1] != b"0")
and body.count(b"\n") >= 3 and body.endswith(b"\n"))
def c2sp_cosignature_verifies(key_bytes, msg, sig_bytes):
"""A witness's hybrid cosignature (K12): both halves over the C2SP message itself, no MoxiVue label."""
if not is_c2sp_cosignature(msg):
return False
k = sig_key(key_bytes)
if isinstance(k, str) or _tag(sig_bytes, TAG_SIG) or len(sig_bytes) != SIG_LEN:
return False
ml_sig = sig_bytes[2 + ED_SIG_LEN:]
if mldsa_sig_decode(ml_sig) is None:
return False
return ed25519_verify_strict(k.ed, msg, sig_bytes[2:2 + ED_SIG_LEN]) and mldsa_verify(k.ml, msg, ml_sig)
def cosignature_from(v):
r = Rd.typed(v, "cosignature")
c = (r.u64("time"), r.bytes("sig"))
r.done()
return c
class Witnesses:
"""Both witnesses' verifier (moxivue-log checkpoint.rs WitnessSet)."""
def __init__(self, slots):
self.slots = slots
@staticmethod
def test(public):
"""the two test witnesses of a test log, by their public keys"""
return Witnesses([{"name": "witness-%d.test" % (i + 1), "keys": [k], "replaced": None}
for i, k in enumerate(public)])
def _counts(self, i, account, size, root, time, sig):
slot = self.slots[i]
m = cosigned_message(account, size, root, time)
if any(c2sp_cosignature_verifies(k, m, sig) for k in slot["keys"]):
return True
r = slot.get("replaced")
if r is not None:
upto = r["upto"].get(account)
if upto is not None and size <= upto[0]:
return any(c2sp_cosignature_verifies(k, m, sig) for k in r["keys"])
return False
def checkpoint(self, account, base, proof):
if self.slots is None:
return False
try:
v = cdec(proof)
if type(v) is not list or len(v) != 2:
return False
c = [cosignature_from(x) for x in v]
except (CanonError, DecodeError):
return False
stamps = (base.t1, base.t2)
return all(c[i][0] == stamps[i] and self._counts(i, account, base.size, base.root, c[i][0], c[i][1])
for i in range(2))
def cancel_list(self, account, l):
if self.slots is None:
return False
m = cenc({"t": "cancel-list-for", "acct": account, "list": l.signed_part})
return any(hybrid_verify(k, L_CANCEL_LIST, m, l.proof) for k in self.slots[l.witness]["keys"])
# ======================================================================== the command
def read_golden(b):
"""(name, witness keys, leaves, checkpoints, failure, refused [(pos, rule)], cosigned) of a stored test log."""
v = cdec(b)
r = Rd.typed(v, "golden")
name = r.text("name")
w = r.array("witnesses")
if len(w) != 2 or not all(type(x) is bytes for x in w):
_bad("witnesses")
leaves, cps = export_from(r.bytes("export"))
vr = Rd.typed(r.get("verdict"), "verdict")
failure = vr.get("failure")
if failure is not None and not _is_u64(failure):
_bad("failure")
refused = []
for x in vr.array("refused"):
if type(x) is not list or len(x) != 2 or not _is_u64(x[0]) or type(x[1]) is not str:
_bad("refused")
refused.append((x[0], x[1]))
cosigned = vr.u64("cosigned")
vr.u64("entries")
vr.done()
r.done()
return name, w, leaves, cps, failure, refused, cosigned
def check_shared(path):
"""The shared signed vectors (crates/moxivue-log/tests/vectors/shared.canon): the verdicts the Rust core gives."""
with open(path, "rb") as f:
v = cdec(f.read())
r = Rd.typed(v, "shared-vectors")
if r.u64("v") != 1:
_bad("unknown version")
n = bad = 0
def differ(what, name):
nonlocal bad
bad += 1
print("DIFFERENT %s: %s" % (what, name))
for c in r.array("signatures"):
n += 1
key_ok = not isinstance(sig_key(c["key"]), str)
ok = hybrid_verify(c["key"], c["label"], c["msg"], c["sig"])
if key_ok != c["key_ok"] or ok != c["valid"]:
differ("signature", c["name"])
for c in r.array("kem_keys"):
n += 1
if (kem_key(c["key"]) is None) != c["ok"]:
differ("X-Wing key", c["name"])
for c in r.array("entries"):
n += 1
try:
sg = signed_from(c["signed"])
except DecodeError:
sg = None
if (sg is not None) != c["ok"]:
differ("entry", c["name"])
continue
if sg is None:
continue
e = sg.entry
same = e.hash == c["hash"] and e.kind == c["kind"]
if e.kind == "propose" and e.b["change"].kind == "join":
tr = e.b["change"].f["t"]
same = same and tr.hash() == c["transcript_hash"] and tr.words() == c["words"]
if not same:
differ("entry", c["name"])
for c in r.array("trees"):
n += 1
tr = Tree()
roots = [tr.roots[0]]
for h in c["leaves"]:
tr.push(h)
roots.append(tr.roots[-1])
if roots != c["roots"]:
differ("tree", c["name"])
print("shared vectors: %d cases, %d different" % (n, bad))
return bad == 0 and n > 0
def first_problem(rep):
if rep.failure is not None:
return rep.failure[0], "verification", rep.failure[1]
if rep.refused:
i, rf = rep.refused[0]
return i, rf.rule, rf.why
return None
def run_export(path, as_json, test_log=False, records=None):
"""The command: verify an export with the built-in witness keys, or (test_log) a stored test log with the test
witnesses' keys it carries (said so in the output; for checking this command, never for a real account)."""
try:
with open(path, "rb") as f:
data = f.read()
rdata = None
if records is not None:
with open(records, "rb") as f:
rdata = f.read()
except OSError as e:
print("%s: %s" % (e.filename, e.strerror), file=sys.stderr)
return 2
keys = WITNESS_KEYS
try:
if test_log:
_, keys, leaves, cps = read_golden(data)[:4]
else:
leaves, cps = export_from(data)
except (DecodeError, CanonError) as e:
rep = None
why = "not a 5.0 log export: %s" % (e.args[0],)
else:
try:
rep = verify(leaves, cps, (Witnesses.test(keys) if test_log else Witnesses(keys)))
except Exception as e: # never fails open: anything unexpected is a refusal
rep = None
why = "the check stopped: %r" % (e,)
if rep is None:
if as_json:
print(json.dumps({"verified": False, "entries": 0, "cosigned": 0, "refused": 0,
"first": {"entry": None, "rule": "verification", "why": why}}, ensure_ascii=False))
else:
print("NOT verified: %s" % why)
return 1
first = first_problem(rep)
ok = rep.clean()
rec = None
if rdata is not None:
# records only against a log that verifies (refused entries have no effect, as for a console)
rec = {"items": 0, "actions": 0, "refused": [], "missing": []}
if rep.failure is None:
try:
items = records_export_from(rdata)
except DecodeError as e:
items = None
rec["refused"].append([None, "format", "not a records export: %s" % e.args[0]])
if items is not None:
rc = RecordCheck(rep.state)
for i, b in enumerate(items):
why = rc.check(b)
if why is not None:
rec["refused"].append([i, why, RECORD_WHY[why]])
rec["items"] = len(items)
rec["actions"] = rc.actions
rec["missing"] = [list(x) for x in rc.missing()]
else:
rec["refused"].append([None, "log", "the log does not verify"])
if rec["refused"]:
ok = False
if first is None:
i, w, text = rec["refused"][0]
first = ("record %s" % i if i is not None else "records", "R6.8", text)
if as_json:
out = {"verified": ok, "entries": rep.entries, "cosigned": rep.cosigned, "refused": len(rep.refused),
"first": None if first is None else {"entry": first[0], "rule": first[1], "why": first[2]},
"refused_entries": [[i, rf.rule] for i, rf in rep.refused],
"account": rep.state.account.hex(),
"records": rec,
"witness_keys": ("TEST keys from the file" if test_log else
"built in" if WITNESS_KEYS is not None else "none yet (5.0 witnesses pending)")}
print(json.dumps(out, ensure_ascii=False))
elif ok:
print("verified from genesis: %d entries, %d in cosigned checkpoints%s"
% (rep.entries, rep.cosigned, " (TEST witnesses from the file)" if test_log else ""))
if rec is not None:
print("records and status: %d, each from a machine of the log with its record key, in its place and in"
" order (not opened: this checker holds no keys)" % (rec["items"] - rec.get("actions", 0)))
if rec.get("actions"):
print("requesters' action records (stored.read): %d, each signed by an active console of the account"
" at a checkpoint of the log" % rec["actions"])
for m, a, b in rec["missing"]:
print(" missing: device %d, record sequences %d to %d" % (m, a, b))
else:
print("NOT verified: %s: %s: %s" % ((first[0] if isinstance(first[0], str) else "entry %d" % first[0]),
first[1], first[2]))
if rec is not None:
for i, w, text in rec["refused"][1:] if rep.clean() else rec["refused"]:
print(" also refused: %s: %s" % ("record %s" % i if i is not None else "records", text))
for i, rf in rep.refused[1:] if rep.failure is None else rep.refused:
print(" also refused: entry %d: %s: %s" % (i, rf.rule, rf.why))
if keys is None and rep.failure is not None and "witnesses" in rep.failure[1]:
print(" (this verifier has no witness keys yet: the 5.0 witnesses' keys are not final)")
return 0 if ok else 1
def golden_files(paths):
out = []
for p in paths:
if os.path.isdir(p):
out += sorted(os.path.join(p, f) for f in os.listdir(p) if f.endswith(".mxgold"))
else:
out.append(p)
return out
def check_golden(paths):
"""Every test log gives the stored verdict: the same failure position, every refused entry with the same rule,
the same cosigned size."""
files = golden_files(paths)
bad = 0
total = 0.0
for p in files:
with open(p, "rb") as f:
name, wk, leaves, cps, failure, refused, cosigned = read_golden(f.read())
t0 = time.time()
rep = verify(leaves, cps, Witnesses.test(wk))
dt = time.time() - t0
total += dt
got_failure = rep.failure[0] if rep.failure is not None else None
got_refused = [(i, rf.rule) for i, rf in rep.refused]
same = got_failure == failure and got_refused == refused and rep.cosigned == cosigned
if not same:
bad += 1
print("DIFFERENT %s: failure %r (stored %r), refused %r (stored %r), cosigned %d (stored %d)%s"
% (name, got_failure, failure, got_refused, refused, rep.cosigned, cosigned,
"; " + rep.failure[1] if rep.failure else ""))
else:
print("same %-80s %4d entries, %d refused, %.2f s" % (name, rep.entries, len(got_refused), dt))
print("test logs: %d, %d different, %.1f s" % (len(files), bad, total))
return bad == 0 and len(files) > 0
_MEMO = {}
def _memo_verify(key_bytes, label, msg, sig_bytes):
k = (key_bytes, label, msg, sig_bytes)
r = _MEMO.get(k)
if r is None:
r = _plain_hybrid_verify(key_bytes, label, msg, sig_bytes)
_MEMO[k] = r
return r
_plain_hybrid_verify = hybrid_verify
def mutate(path, per_leaf):
"""Changes single bytes of a test log (one at a time; positions sampled across every leaf: the leaf's head,
the signed entry, its signatures, and every checkpoint) and checks that each changed log is not verified.
Signature results are memoised (a pure function) so the unchanged part is not checked again."""
global hybrid_verify
hybrid_verify = _memo_verify
with open(path, "rb") as f:
name, wk, leaves, cps, failure, refused, cosigned = read_golden(f.read())
w = Witnesses.test(wk)
base = verify(leaves, cps, w)
if not base.clean():
print("%s: the unchanged log is not verified; mutations need a verified log" % name)
return False
tried = accepted = 0
for i, leaf in enumerate(leaves):
n = len(leaf)
offsets = sorted(set([0, 1, 2, n // 7, n // 3, n // 2, (2 * n) // 3, n - SIG_LEN // 2, n - 3, n - 1] +
[((k * 7919 + i * 104729) % n) for k in range(per_leaf)]))
for o in offsets:
if not 0 <= o < n:
continue
m = bytearray(leaf)
m[o] ^= 1 << ((o + i) % 8)
ls = list(leaves)
ls[i] = bytes(m)
rep = verify(ls, cps, w)
tried += 1
if rep.clean():
accepted += 1
print("ACCEPTED: %s leaf %d byte %d changed" % (name, i, o))
# whole leaves: two neighbours swapped, one repeated, one in the middle dropped
for i in range(len(leaves) - 1):
for ls in (leaves[:i] + [leaves[i + 1], leaves[i]] + leaves[i + 2:], leaves[:i + 1] + [leaves[i]] + leaves[i + 1:],
leaves[:i] + leaves[i + 1:]):
tried += 1
if verify(ls, cps, w).clean():
accepted += 1
print("ACCEPTED: %s leaves reordered, repeated or dropped at %d" % (name, i))
for j, (b, proof) in enumerate(cps):
for o in sorted(set([0, len(proof) // 2, len(proof) - 1])):
m = bytearray(proof)
m[o] ^= 0x10
c2 = list(cps)
c2[j] = (b, bytes(m))
tried += 1
if verify(leaves, c2, w).clean():
accepted += 1
print("ACCEPTED: %s checkpoint %d byte %d changed" % (name, j, o))
for fld in range(4):
bb = list(b)
bb[fld] = bb[fld][:-1] + bytes([bb[fld][-1] ^ 1]) if fld == 1 else bb[fld] + 1
c2 = list(cps)
c2[j] = (Base(*bb), proof)
tried += 1
if verify(leaves, c2, w).clean():
accepted += 1
print("ACCEPTED: %s checkpoint %d field %d changed" % (name, j, fld))
hybrid_verify = _plain_hybrid_verify
print("%s: %d changes (single bytes of every leaf and checkpoint, whole leaves), %d accepted" % (name, tried, accepted))
return accepted == 0 and tried > 0
def speed(path):
with open(path, "rb") as f:
name, wk, leaves, cps, failure, refused, cosigned = read_golden(f.read())
counter = [0]
def counting(*a):
counter[0] += 1
return _plain_hybrid_verify(*a)
global hybrid_verify
hybrid_verify = counting
t0 = time.time()
rep = verify(leaves, cps, Witnesses.test(wk))
dt = time.time() - t0
hybrid_verify = _plain_hybrid_verify
print("%s: %d entries, %d checkpoints, %d hybrid signatures checked, %.2f s (%.1f ms an entry, %.1f ms a "
"signature)" % (name, rep.entries, len(cps), counter[0], dt, 1000 * dt / max(1, rep.entries),
1000 * dt / max(1, counter[0])))
return True
def self_test(vdir):
"""The published vectors (the same files moxivue-proto's tests read), with the counts the Rust tests expect."""
def load(rel):
with open(os.path.join(vdir, rel), encoding="utf-8") as f:
return json.load(f)
def wy(v):
for g in v["testGroups"]:
for t in g["tests"]:
yield g, t
ok = True
def report(what, good, detail=""):
nonlocal ok
ok = ok and good
print("%s %s%s" % ("pass" if good else "FAIL", what, detail))
# Ed25519: Wycheproof through the key check and strict verification
valid = invalid = wrong = 0
for g, t in wy(load("wycheproof/ed25519_test.json")):
k = accept_ed25519(bytes.fromhex(g["publicKey"]["pk"]))
if isinstance(k, str):
wrong += 1
continue
r = ed25519_verify_strict(k, bytes.fromhex(t["msg"]), bytes.fromhex(t["sig"]))
if t["result"] == "valid":
valid += r
wrong += not r
else:
invalid += not r
wrong += r
report("Ed25519 Wycheproof", (valid, invalid, wrong) == (88, 63, 0), ": %d valid, %d invalid" % (valid, invalid))
# ed25519-speccheck: none of the 12 edge cases is accepted (small and mixed-order keys refused, R2.4)
why = []
for c in load("speccheck/cases.json"):
k = accept_ed25519(bytes.fromhex(c["pub_key"]))
if isinstance(k, str):
why.append(k)
elif ed25519_verify_strict(k, bytes.fromhex(c["message"]), bytes.fromhex(c["signature"])):
why.append("ACCEPTED")
else:
why.append("signature refused")
report("ed25519-speccheck", len(why) == 12 and "ACCEPTED" not in why, ": " + ", ".join(why))
# X25519 halves: low order, non-canonical and twist keys refused
bad = zero = 0
for g, t in wy(load("wycheproof/x25519_test.json")):
u = bytes.fromhex(t["public"])
r = accept_x25519(u)
flags = t.get("flags", [])
if t["shared"] == "00" * 32:
zero += 1
bad += r not in ("X25519LowOrder", "X25519NonCanonical")
if any(f in flags for f in ("Twist", "NonCanonicalPublic", "LowOrderPublic")):
bad += r is None
if r is not None and t["result"] == "valid":
bad += r not in ("X25519NonCanonical", "X25519Twist")
report("X25519 Wycheproof classification", bad == 0 and zero >= 30, ": %d zero-secret keys refused" % zero)
# ML-KEM-768 modulus check
valid = invalid = wrong = 0
for g, t in wy(load("wycheproof/mlkem_768_encaps_test.json")):
r = accept_mlkem768(bytes.fromhex(t["ek"]))
if t["result"] == "valid":
valid += r is None
wrong += r is not None
else:
invalid += r is not None
wrong += r is None
report("ML-KEM-768 modulus check (Wycheproof)", (valid, invalid, wrong) == (133, 132, 0),
": %d valid, %d invalid" % (valid, invalid))
# ML-DSA-65: Wycheproof verify (incl. invalid hint encodings) and NIST ACVP sigVer
table = {}
wrong = 0
good_sigs = []
for g, t in wy(load("wycheproof/mldsa_65_verify_test.json")):
pk = bytes.fromhex(g["publicKey"])
if len(pk) != MLDSA_VK_LEN:
r = False
else:
k = MlDsaKey(pk)
msg, sig = bytes.fromhex(t["msg"]), bytes.fromhex(t["sig"])
r = mldsa_verify(k, msg, sig, bytes.fromhex(t.get("ctx") or ""))
if r and not t.get("ctx"):
good_sigs.append((k, msg, sig))
want = t["result"] == "valid"
wrong += r != want
table[(t["result"], r)] = table.get((t["result"], r), 0) + 1
report("ML-DSA-65 Wycheproof verify", wrong == 0 and sum(table.values()) == 210, ": %r" % sorted(table.items()))
passed = refused = wrong = differ = 0
for g, t in wy(load("acvp/ML-DSA-sigVer-FIPS204-65.json")):
k = MlDsaKey(bytes.fromhex(t["pk"]))
sig = bytes.fromhex(t["signature"])
if g["externalMu"]:
mu = bytes.fromhex(t["mu"])
elif g["signatureInterface"] == "internal":
mu = hashlib.shake_256(k.tr + bytes.fromhex(t["message"])).digest(64)
else:
ctx = bytes.fromhex(t["context"])
mu = hashlib.shake_256(k.tr + bytes([0, len(ctx)]) + ctx + bytes.fromhex(t["message"])).digest(64)
r = mldsa_verify_mu(k, mu, sig)
differ += r != mldsa_verify_mu(k, mu, sig, reference=True)
wrong += r != t["testPassed"]
passed += r
refused += not r
report("ML-DSA-65 ACVP sigVer", wrong == 0 and differ == 0 and passed + refused == 45,
": %d valid, %d refused; the transform path agrees on all" % (passed, refused))
# repeated and unordered hint indices (CVE-2026-24850): the same hint in another encoding is refused
dup = swapped = accepted = lax = 0
def _lax_verify(k, msg, sig):
# what a verifier without Algorithm 21's order check makes of it: the same hint vector, so it verifies
d = mldsa_sig_decode(sig[:CT_LEN + ML_ * 640] + bytes(OMEGA + MK))
hs = CT_LEN + ML_ * 640
h = [[0] * 256 for _ in range(MK)]
start = 0
for i in range(MK):
end = sig[hs + OMEGA + i]
for x in sig[hs + start:hs + end]:
h[i][x] = 1
start = end
mu = hashlib.shake_256(k.tr + bytes([0, 0]) + msg).digest(64)
return _mldsa_check(k, mu, (d[0], d[1], h))
for k, msg, sig in good_sigs:
hs = len(sig) - (OMEGA + MK)
counts = list(sig[hs + OMEGA:])
total = counts[-1]
prev = 0
j = None
for x, c in enumerate(counts):
if c > prev:
j = x
break
prev = c
if total < OMEGA and j is not None:
pos = counts[j] - 1
idx = list(sig[hs:hs + total])
idx.insert(pos + 1, idx[pos])
bad = bytearray(sig)
bad[hs:hs + total + 1] = bytes(idx)
for x in range(j, MK):
bad[hs + OMEGA + x] += 1
dup += 1
accepted += mldsa_verify(k, msg, bytes(bad))
lax += _lax_verify(k, msg, bytes(bad))
start = 0
for end in counts:
if end >= start + 2:
bad = bytearray(sig)
bad[hs + start], bad[hs + start + 1] = bad[hs + start + 1], bad[hs + start]
swapped += 1
accepted += mldsa_verify(k, msg, bytes(bad))
lax += _lax_verify(k, msg, bytes(bad))
break
start = end
report("ML-DSA-65 repeated and unordered hint indices refused", accepted == 0 and dup >= 20 and swapped >= 20
and lax == dup + swapped, ": %d repeated, %d swapped, none accepted (all %d verify without the order "
"check)" % (dup, swapped, lax))
# the two multiplication paths agree on random polynomials
import random
rnd = random.Random(5)
agree = True
for _ in range(5):
a = [rnd.randrange(MQ) for _ in range(256)]
b = [rnd.randrange(MQ) for _ in range(256)]
via_ntt = intt([x * y % MQ for x, y in zip(ntt(a), ntt(b))])
agree = agree and via_ntt == _unpack_fold(_pack(a) * _pack(b)) and intt(ntt(a)) == a
report("negacyclic products: transform and big-integer paths agree", agree)
return ok
USAGE = """usage: python3 verify-audit5.py <export> [--records <records-export>] [--json]
python3 verify-audit5.py --test-log <golden-file> [--records <records-export>] [--json]
python3 verify-audit5.py --self-test <vectors-dir>
python3 verify-audit5.py --golden <file-or-dir>...
python3 verify-audit5.py --shared <shared.canon>
python3 verify-audit5.py --records-vectors <records.canon>
python3 verify-audit5.py --mutate <golden-file> [samples-per-leaf]
python3 verify-audit5.py --speed <golden-file>
exit codes: 0 verified (or every check passed), 1 not verified (or a check failed), 2 usage"""
def main(argv):
args = argv[1:]
if not args or args[0] in ("-h", "--help"):
print(USAGE, file=sys.stderr)
return 2
try:
if args[0] == "--self-test" and len(args) == 2:
return 0 if self_test(args[1]) else 1
if args[0] == "--golden" and len(args) >= 2:
return 0 if check_golden(args[1:]) else 1
if args[0] == "--records-vectors" and len(args) == 2:
return 0 if check_records_vectors(args[1]) else 1
if args[0] == "--shared" and len(args) == 2:
return 0 if check_shared(args[1]) else 1
if args[0] == "--mutate" and len(args) in (2, 3):
n = int(args[2]) if len(args) == 3 else 6
return 0 if mutate(args[1], n) else 1
if args[0] == "--speed" and len(args) == 2:
return 0 if speed(args[1]) else 1
except (OSError, DecodeError, CanonError, ValueError, KeyError) as e:
print("error: %r" % (e,), file=sys.stderr)
return 1
as_json = "--json" in args
test_log = "--test-log" in args
records = None
rest = [a for a in args if a not in ("--json", "--test-log")]
if "--records" in rest:
i = rest.index("--records")
if i + 1 >= len(rest):
print(USAGE, file=sys.stderr)
return 2
records = rest[i + 1]
del rest[i:i + 2]
if len(rest) != 1 or rest[0].startswith("--") or len(args) != len(set(args)):
print(USAGE, file=sys.stderr)
return 2
return run_export(rest[0], as_json, test_log, records)
if __name__ == "__main__":
sys.exit(main(sys.argv))