fdd0b72b7a
The complement to the per-function compare harness. Instead of "did this function's declared outputs match", it asks "is the entire simulation state still identical" -- so no region-declaration mistake can hide from it. Coverage is PROVED, not declared: the digest tree is re-serialised and compared byte-for-byte against the inflated save on every run. When that reconstruction reproduces the stream, the whole file is a function of the digest's inputs. A run that cannot account for the file says so and exits non-zero. This is the direct answer to B4's three hooks that printed "0 diverged" over an empty region set. It localises. The root is the fold of a per-subsystem / per-object tree with named objects, so the known load->re-save delta reports as exactly five leaves -- /Summary/Checksum and four /Sim/players/Player[...]/Status 4->0 -- naming the two Singularity players by id where the raw byte diff could only say "1st of two". One real End Turn reports as 108 fully attributed differences. Float-parity policy is explicit and strict by default (STATE_CHECKSUM.md 3): raw IEEE-754 bits; a `canonical` policy for signed zero and NaN payloads only; and deliberately NO tolerant hashing mode, because quantisation moves the cliff rather than removing it and destroys the roll-up. Tolerance lives in the differ as --ulps, applied after localisation. Corpus census: 0 NaN, 0 -0.0, 0 subnormals across 4,474 float leaves, so the strict default costs nothing today and a test fails the day that changes. Validated on the real saves (verify/results/state-checksum/): 10 files, 4 distinct contents, all STABLE + COVERED; chain record/verify works on the real turn1-3 saves. The VM-driven replay loop is designed (section 5) but UNRUN. Section 3.5 names the one question the host side cannot settle -- whether the turn pipeline depends on x87 intermediate precision -- and the experiment that would: force fpu_cw to 0x027f / 0x127f / 0x137f across End Turn and checksum the three autosaves. Also recorded: Summary.Checksum is NOT a byte sum over the inflated stream nor a sum over the int leaves (both ruled out), so nobody repeats those two. 38 tests; sots-engine untouched, clean_room_check.sh OK.
846 lines
34 KiB
Python
846 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""state_checksum.py -- whole-state, diagnostic checksum of a SOTS1 save.
|
|
|
|
Why this exists
|
|
---------------
|
|
Every other verification in this project is *per-function*: hook one routine,
|
|
compare the regions it declares, print a verdict. That is only as good as the
|
|
region declaration, and region declarations have been wrong (B4: three hooks
|
|
that reported "0 diverged" while comparing nothing; the harness audit: 23
|
|
undeclared side effects). A green verdict over an empty region set is a
|
|
failure, not a pass.
|
|
|
|
This tool is the complement. It does not ask "did the declared outputs match".
|
|
It asks **"is the entire simulation state still identical"**, computed from the
|
|
one artefact the game emits as a pure function of state -- the save file, which
|
|
`findings/subsystems/determinism-oracle.md` proved byte-identical across runs
|
|
and processes.
|
|
|
|
Two properties make it evidence rather than a comforting number:
|
|
|
|
1. **Provable coverage.** The checksum is computed from a tree that is
|
|
re-serialised and compared byte-for-byte against the inflated save
|
|
(`--audit`, on by default). If the reconstruction reproduces the stream,
|
|
then every byte of the save is a function of the checksum's inputs, so *no*
|
|
state change can be invisible to it. Coverage is asserted by construction,
|
|
not by a hand-written region list that someone can forget to fill in.
|
|
2. **It localises.** The root digest is the fold of a tree of per-subsystem /
|
|
per-object digests, and objects are named (`Player[16 "re"]`,
|
|
`Sys[112 "Gamma Cephei"]`, `Events/[turn=3]/[id=3]`), so a divergence
|
|
reports *which object moved*, not just that the hash moved.
|
|
|
|
Float parity: see STATE_CHECKSUM.md section 3. Short version -- the digest is
|
|
always exact (raw IEEE-754 bits, or bits with -0.0/NaN normalised under
|
|
`--floats canonical`). Tolerance is deliberately *not* a hashing mode; it
|
|
exists only in the differ (`--ulps`), because a tolerant hash is a
|
|
contradiction: quantisation just moves the cliff, it does not remove it.
|
|
|
|
Usage
|
|
-----
|
|
state_checksum.py SAVE # root + subsystem digests
|
|
state_checksum.py SAVE --tree --depth 3 # digest tree
|
|
state_checksum.py A B # localise divergence
|
|
state_checksum.py A B --ulps 2 # classify float diffs
|
|
state_checksum.py --chain chain.json S1 S2 ... # verify a recorded chain
|
|
state_checksum.py SAVE --json # machine-readable
|
|
|
|
Stdlib only (plus save_reader.py, which is stdlib only).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import struct
|
|
import sys
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_READER = os.path.join(os.path.dirname(_HERE), "save-reader")
|
|
if _READER not in sys.path:
|
|
sys.path.insert(0, _READER)
|
|
|
|
import save_reader as sr # noqa: E402
|
|
|
|
__all__ = [
|
|
"FLOAT_POLICIES", "MASK_PRESETS", "CkNode", "checksum_save", "checksum_bytes",
|
|
"reconstruct", "audit_coverage", "diff", "DiffEntry", "float_key",
|
|
]
|
|
|
|
DIGEST_BYTES = 16 # blake2b-128; 2**-64 collision floor at our object counts
|
|
SHORT = 16 # hex chars shown in text output
|
|
|
|
|
|
def _reader_fingerprint() -> str:
|
|
"""Identity of the save_reader build the digests were computed under.
|
|
|
|
The digest hashes each leaf's *inferred kind* alongside its bytes, so it is
|
|
a function of (save bytes, reader schema) -- not of the bytes alone. That
|
|
is fine for comparing two saves parsed by one reader, and dangerous for a
|
|
chain recorded months ago. So it is recorded and checked, never assumed.
|
|
It is deliberately NOT folded into the digest: a cosmetic reader edit
|
|
should not invalidate every recorded root, it should raise a warning.
|
|
"""
|
|
try:
|
|
with open(sr.__file__, "rb") as f:
|
|
return hashlib.blake2b(f.read(), digest_size=8).hexdigest()
|
|
except OSError: # pragma: no cover
|
|
return "unknown"
|
|
|
|
|
|
READER_FINGERPRINT = _reader_fingerprint()
|
|
|
|
BEGIN = struct.pack("<I", 0xBEEFBEEF)
|
|
END = struct.pack("<I", 0x41104110)
|
|
|
|
# --- policy -------------------------------------------------------------------
|
|
|
|
FLOAT_POLICIES = ("bits", "canonical")
|
|
|
|
#: Documented canonicalisation presets (see determinism-oracle.md). A rule is
|
|
#: (leaf tag, required path prefix, replacement kind, replacement value) -- the
|
|
#: path prefix keeps a mask from silently swallowing a same-named field
|
|
#: somewhere else in the tree, and the run reports how many leaves each rule
|
|
#: actually hit so an over-broad mask is visible rather than comfortable.
|
|
#:
|
|
#: Masking is opt-in. By default nothing is masked, so the known re-save delta
|
|
#: is *localised* rather than absorbed.
|
|
MASK_PRESETS = {
|
|
"none": (),
|
|
# load -> save of a post-turn autosave: the server clears the per-player
|
|
# "turn ended" flag (4 -> 0) and the derived top-level Checksum follows.
|
|
"resave": (
|
|
("Status", "/Sim/players/Player[", "int", 0),
|
|
("Checksum", "/Summary/", "derived", None),
|
|
),
|
|
}
|
|
|
|
|
|
def float_key(raw: bytes, policy: str) -> bytes:
|
|
"""Hashing key for a 4-byte IEEE-754 single under `policy`."""
|
|
if policy == "bits":
|
|
return raw
|
|
if policy != "canonical":
|
|
raise ValueError(f"unknown float policy {policy!r}")
|
|
(bits,) = struct.unpack("<I", raw)
|
|
if bits == 0x80000000: # -0.0 -> +0.0
|
|
return b"\0\0\0\0"
|
|
if (bits & 0x7F800000) == 0x7F800000 and (bits & 0x007FFFFF):
|
|
return struct.pack("<I", 0x7FC00000) # any NaN -> one quiet NaN
|
|
return raw
|
|
|
|
|
|
def ulps_apart(a: bytes, b: bytes) -> float:
|
|
"""Distance in float32 ULPs between two raw singles. inf if not comparable."""
|
|
(x,) = struct.unpack("<f", a)
|
|
(y,) = struct.unpack("<f", b)
|
|
if math.isnan(x) or math.isnan(y):
|
|
return 0.0 if (math.isnan(x) and math.isnan(y)) else math.inf
|
|
if math.isinf(x) or math.isinf(y):
|
|
return 0.0 if x == y else math.inf
|
|
|
|
def ordinal(v: bytes) -> int:
|
|
(u,) = struct.unpack("<i", v)
|
|
return u if u >= 0 else -0x80000000 - u # monotone map across the sign
|
|
return float(abs(ordinal(a) - ordinal(b)))
|
|
|
|
|
|
# --- naming -------------------------------------------------------------------
|
|
|
|
#: lead-int tag -> (frame tag, group name). The save stores object tables as a
|
|
#: flat run of (id, frame) sibling pairs; folding each run into a named group
|
|
#: turns `Sim`'s 259 children into a subsystem tree.
|
|
PAIR_GROUPS = {
|
|
"PlayerID": ("Player", "players"),
|
|
"SysID": ("Sys", "systems"),
|
|
"FltID": ("Flt", "fleets"),
|
|
"ShipID": ("Ship", "ships"),
|
|
"DesID": ("Des", "designs"),
|
|
"TradeID": ("Trade", "trades"),
|
|
"ply": ("hist", "history"),
|
|
}
|
|
|
|
#: frame tag -> child tags to use as a human label, first match wins. Tags are
|
|
#: as spelled on disk (SAVE_FORMAT.md section 10); `Ship` carries no name field,
|
|
#: so a ship is labelled by its id alone.
|
|
NAME_FIELDS = {
|
|
"Player": ("PlryName",),
|
|
"Sys": ("Name",),
|
|
"Flt": ("FtName",),
|
|
"Des": ("DName",),
|
|
}
|
|
|
|
#: child tags that identify a NULL-named ('.') element frame
|
|
ELEM_KEYS = ("EvTurn", "EvEID", "ordID", "npid", "desID", "Wpt", "seno2", "ID", "set")
|
|
|
|
#: NonComplexArray count tags whose elements are a run of `.` scalars in the
|
|
#: *same* frame (SAVE_FORMAT.md section 4, "inline count"). Left flat, a single
|
|
#: id insertion shifts every following sibling and the differ reports a hundred
|
|
#: spurious moves; folded, it reports one list that gained one id.
|
|
INLINE_ID_LISTS = ("PlayerIDs", "DesignIDs", "SystemIDs", "FleetIDs",
|
|
"ShipIDs", "TradeIDs")
|
|
|
|
#: top-level digest rows printed by the default (non-tree) view. Paths that do
|
|
#: not exist in a given save are silently skipped.
|
|
SUBSYSTEM_PATHS = (
|
|
"/Summary",
|
|
"/CreateParams",
|
|
"/Sim/RNG",
|
|
"/Sim/Attrib",
|
|
"/Sim/turnstats",
|
|
"/Sim/players",
|
|
"/Sim/systems",
|
|
"/Sim/fleets",
|
|
"/Sim/NdGr2",
|
|
"/Sim/trdmgr",
|
|
"/Sim/spymgr",
|
|
"/Sim/SvSctOb",
|
|
"/CDT",
|
|
)
|
|
|
|
#: per-player sub-rollups, appended under each player row in --subsystems
|
|
PLAYER_PARTS = ("TechTree", "Events", "designs", "Diplo", "spy2", "civr", "comms", "Ojvs")
|
|
|
|
|
|
# --- the digest tree ----------------------------------------------------------
|
|
|
|
class CkNode:
|
|
"""One node of the checksum tree."""
|
|
|
|
__slots__ = ("path", "label", "kind", "digest", "offset", "size",
|
|
"leaves", "value_bytes", "children", "value", "raw", "masked",
|
|
"values")
|
|
|
|
def __init__(self, path, label, kind, digest, offset, size,
|
|
leaves, value_bytes, children, value=None, raw=b"", masked=False,
|
|
values=None):
|
|
self.values = values # inline id list contents, for set-diffing
|
|
self.path = path
|
|
self.label = label
|
|
self.kind = kind
|
|
self.digest = digest
|
|
self.offset = offset
|
|
self.size = size
|
|
self.leaves = leaves # number of scalar leaves below (or 1)
|
|
self.value_bytes = value_bytes # bytes of *value* payload below (framing excluded)
|
|
self.children = children
|
|
self.value = value
|
|
self.raw = raw
|
|
self.masked = masked
|
|
|
|
@property
|
|
def hexd(self) -> str:
|
|
return self.digest.hex()
|
|
|
|
def short(self) -> str:
|
|
return self.digest.hex()[:SHORT]
|
|
|
|
def find(self, path: str):
|
|
if self.path == path:
|
|
return self
|
|
for c in self.children:
|
|
if path == c.path or path.startswith(c.path + "/"):
|
|
return c.find(path)
|
|
return None
|
|
|
|
def walk(self):
|
|
yield self
|
|
for c in self.children:
|
|
yield from c.walk()
|
|
|
|
def as_dict(self, depth: int = 99, _d: int = 0) -> dict:
|
|
out = {"path": self.path, "kind": self.kind, "digest": self.hexd,
|
|
"offset": self.offset, "size": self.size,
|
|
"leaves": self.leaves, "valueBytes": self.value_bytes}
|
|
if self.masked:
|
|
out["masked"] = True
|
|
if self.kind != "complex" and self.value is not None:
|
|
out["value"] = self.value
|
|
if self.children and _d < depth:
|
|
out["children"] = [c.as_dict(depth, _d + 1) for c in self.children]
|
|
elif self.children:
|
|
out["childCount"] = len(self.children)
|
|
return out
|
|
|
|
|
|
def _h(*parts: bytes) -> bytes:
|
|
d = hashlib.blake2b(digest_size=DIGEST_BYTES)
|
|
for p in parts:
|
|
d.update(struct.pack("<I", len(p)))
|
|
d.update(p)
|
|
return d.digest()
|
|
|
|
|
|
def _label(node: sr.Node, index: int, lead: sr.Node | None,
|
|
unique: bool = False) -> str:
|
|
"""Human, stable, greppable label for one item.
|
|
|
|
A key is appended only when it carries information: an object id, an object
|
|
name, or an element key. A bare positional index is appended only when the
|
|
tag is ambiguous among its siblings (`unique=False`), so a uniquely-named
|
|
field reads as `Status`, not `Status[36]`.
|
|
"""
|
|
base = node.name if node.name is not None else "<frame>"
|
|
key = None
|
|
if lead is not None and isinstance(lead.value, int):
|
|
key = str(lead.value)
|
|
if node.kind == "complex":
|
|
want = NAME_FIELDS.get(node.name or "")
|
|
if want:
|
|
for c in node.children:
|
|
if c.name in want and isinstance(c.value, str):
|
|
key = f'{key} "{c.value}"' if key else f'"{c.value}"'
|
|
break
|
|
if key is None and (node.name in (".", None)):
|
|
for c in node.children:
|
|
if c.name in ELEM_KEYS:
|
|
key = f"{c.name}={c.value}"
|
|
break
|
|
if key is None:
|
|
if unique:
|
|
return base
|
|
key = str(index)
|
|
return f"{base}[{key}]"
|
|
|
|
|
|
def _name_counts(nodes) -> dict:
|
|
counts: dict = {}
|
|
for n in nodes:
|
|
counts[n.name] = counts.get(n.name, 0) + 1
|
|
return counts
|
|
|
|
|
|
def _group_children(nodes: list[sr.Node]):
|
|
"""Fold object runs and inline id lists into synthetic groups.
|
|
|
|
Returns a list of entries:
|
|
("item", node, None) -- a plain child
|
|
("group", name, [(id_node, frame_node), ...]) -- an object table
|
|
("list", name, count_node, [value_nodes]) -- an inline id list
|
|
Order and byte coverage are preserved exactly; nothing is dropped, so the
|
|
reconstruction audit still proves total coverage.
|
|
"""
|
|
out, i, seen = [], 0, {}
|
|
n = len(nodes)
|
|
while i < n:
|
|
cur = nodes[i]
|
|
spec = PAIR_GROUPS.get(cur.name or "")
|
|
if (spec and i + 1 < n and nodes[i + 1].name == spec[0]
|
|
and nodes[i + 1].kind == "complex" and cur.kind in ("int", "float")):
|
|
frame_tag, gname = spec
|
|
pairs = []
|
|
while (i + 1 < n and nodes[i].name == cur.name
|
|
and nodes[i + 1].name == frame_tag and nodes[i + 1].kind == "complex"):
|
|
pairs.append((nodes[i], nodes[i + 1]))
|
|
i += 2
|
|
seen[gname] = seen.get(gname, 0) + 1
|
|
label = gname if seen[gname] == 1 else f"{gname}({seen[gname]})"
|
|
out.append(("group", label, pairs))
|
|
continue
|
|
if (cur.name in INLINE_ID_LISTS and cur.kind == "int"
|
|
and isinstance(cur.value, int) and 0 <= cur.value <= n - i - 1
|
|
and all(nodes[i + 1 + k].name == "." and nodes[i + 1 + k].kind != "complex"
|
|
for k in range(cur.value))):
|
|
vals = nodes[i + 1:i + 1 + cur.value]
|
|
out.append(("list", cur.name, cur, vals))
|
|
i += 1 + cur.value
|
|
continue
|
|
out.append(("item", cur, None))
|
|
i += 1
|
|
return out
|
|
|
|
|
|
def _value_key(node: sr.Node, path: str, policy: str, masks, hits: dict):
|
|
"""(kind, key_bytes, masked) for a scalar/raw leaf."""
|
|
kind = node.kind
|
|
for tag, under, mkind, mval in masks:
|
|
if node.name != tag or not path.startswith(under):
|
|
continue
|
|
hits[tag] = hits.get(tag, 0) + 1
|
|
if mkind == "derived":
|
|
return kind, b"<derived>", True
|
|
if mkind == "int":
|
|
return kind, struct.pack("<i", mval), True
|
|
raise ValueError(f"bad mask spec {(tag, under, mkind, mval)!r}")
|
|
raw = node.raw
|
|
if kind == "float":
|
|
return kind, float_key(raw, policy), False
|
|
if kind == "bool" and policy == "canonical":
|
|
return kind, b"\x01" if any(raw) else b"\x00", False
|
|
return kind, raw, False
|
|
|
|
|
|
def _build(node: sr.Node, path: str, label: str, policy: str, masks, hits: dict) -> CkNode:
|
|
if node.kind == "complex":
|
|
entries = _group_children(node.children)
|
|
counts = _name_counts(e[1] for e in entries if e[0] == "item")
|
|
# positional index is the ordinal among *same-named* siblings, so one
|
|
# insertion elsewhere in the frame does not renumber everything after it
|
|
ordinals: dict = {}
|
|
kids: list[CkNode] = []
|
|
for ent in entries:
|
|
if ent[0] == "item":
|
|
child = ent[1]
|
|
k = ordinals.get(child.name, 0)
|
|
ordinals[child.name] = k + 1
|
|
lab = _label(child, k, None, unique=counts.get(child.name) == 1)
|
|
kids.append(_build(child, f"{path}/{lab}", lab, policy, masks, hits))
|
|
elif ent[0] == "list":
|
|
_, lname, cnode, vals = ent
|
|
lpath = f"{path}/{lname}[]"
|
|
keys = [_value_key(v, lpath, policy, masks, hits)[1] for v in vals]
|
|
ldig = _h(b"list", lname.encode("ascii"), cnode.raw, *keys)
|
|
off = cnode.offset
|
|
last = vals[-1] if vals else cnode
|
|
kids.append(CkNode(lpath, f"{lname}[]", "list", ldig, off,
|
|
last.offset + last.size - off,
|
|
1 + len(vals),
|
|
len(cnode.raw) + sum(len(v.raw) for v in vals),
|
|
[], values=[v.value for v in vals]))
|
|
else:
|
|
_, gname, pairs = ent
|
|
gpath = f"{path}/{gname}"
|
|
gkids = []
|
|
for j, (idn, frame) in enumerate(pairs):
|
|
lab = _label(frame, j, idn)
|
|
ck = _build(frame, f"{gpath}/{lab}", lab, policy, masks, hits)
|
|
# the id int is part of the object's identity; fold it in
|
|
idk = _h(b"id", (idn.name or "").encode("ascii"), idn.raw)
|
|
ck.digest = _h(b"obj", idk, ck.digest)
|
|
ck.offset = idn.offset
|
|
ck.size = frame.offset + frame.size - idn.offset
|
|
ck.value_bytes += len(idn.raw)
|
|
ck.leaves += 1
|
|
gkids.append(ck)
|
|
gdig = _h(b"group", gname.encode("ascii"),
|
|
*[k.digest for k in gkids])
|
|
off = gkids[0].offset if gkids else node.offset
|
|
size = (gkids[-1].offset + gkids[-1].size - off) if gkids else 0
|
|
kids.append(CkNode(gpath, gname, "group", gdig, off, size,
|
|
sum(k.leaves for k in gkids),
|
|
sum(k.value_bytes for k in gkids), gkids))
|
|
dig = _h(b"frame", (node.name or "").encode("ascii"), *[k.digest for k in kids])
|
|
return CkNode(path, label, "complex", dig, node.offset, node.size,
|
|
sum(k.leaves for k in kids) or 0,
|
|
sum(k.value_bytes for k in kids), kids)
|
|
|
|
kind, key, masked = _value_key(node, path, policy, masks, hits)
|
|
dig = _h(b"leaf", (node.name or "").encode("ascii"), kind.encode("ascii"), key)
|
|
return CkNode(path, label, kind, dig, node.offset, node.size, 1, len(node.raw),
|
|
[], value=node.value, raw=node.raw, masked=masked)
|
|
|
|
|
|
# --- coverage proof -----------------------------------------------------------
|
|
|
|
def reconstruct(node: sr.Node, out: bytearray, base: int) -> None:
|
|
"""Re-serialise a reader Node back into `out`.
|
|
|
|
`base` is the inflated offset `out` starts at, so joint padding (which is
|
|
computed against the absolute 4-byte grid) lands where it did originally.
|
|
"""
|
|
def pad():
|
|
while (base + len(out)) & 3:
|
|
out.append(0)
|
|
|
|
if node.name is None and node.kind == "raw":
|
|
out += node.raw # unnamed payload / stray bytes
|
|
return
|
|
if node.kind == "complex":
|
|
if node.name is not None:
|
|
b = node.name.encode("ascii")
|
|
out += struct.pack("<i", len(b)) + b
|
|
pad()
|
|
out += BEGIN
|
|
for c in node.children:
|
|
reconstruct(c, out, base)
|
|
out += END
|
|
return
|
|
b = (node.name or "").encode("ascii")
|
|
out += struct.pack("<i", len(b)) + b
|
|
if node.kind == "string":
|
|
out += struct.pack("<i", len(node.raw)) # reader's raw excludes the length prefix
|
|
out += node.raw
|
|
pad()
|
|
|
|
|
|
def audit_coverage(tree: sr.Node, inflated: bytes) -> dict:
|
|
"""Re-serialise the whole tree and compare with the inflated stream."""
|
|
out = bytearray()
|
|
for c in tree.children:
|
|
reconstruct(c, out, 0)
|
|
ok = bytes(out) == inflated
|
|
first = None
|
|
if not ok:
|
|
n = min(len(out), len(inflated))
|
|
for i in range(n):
|
|
if out[i] != inflated[i]:
|
|
first = i
|
|
break
|
|
if first is None:
|
|
first = n
|
|
return {"ok": ok, "rebuiltBytes": len(out), "inflatedBytes": len(inflated),
|
|
"firstDiff": first}
|
|
|
|
|
|
# --- top level ----------------------------------------------------------------
|
|
|
|
class SaveChecksum:
|
|
def __init__(self, path, root: CkNode, coverage: dict, res: sr.SaveResult,
|
|
policy: str, mask: str, mask_hits: dict | None = None):
|
|
self.path = path
|
|
self.root = root
|
|
self.coverage = coverage
|
|
self.res = res
|
|
self.policy = policy
|
|
self.mask = mask
|
|
#: leaves each mask rule actually replaced. Printed on every masked run
|
|
#: so an over-broad mask is visible rather than comfortable.
|
|
self.mask_hits = mask_hits or {}
|
|
|
|
@property
|
|
def digest(self) -> str:
|
|
return self.root.hexd
|
|
|
|
def subsystems(self) -> list[tuple[str, CkNode]]:
|
|
rows = []
|
|
for p in SUBSYSTEM_PATHS:
|
|
n = self._by_tag(p)
|
|
if n is not None:
|
|
rows.append((p, n))
|
|
return rows
|
|
|
|
def _by_tag(self, tagpath: str):
|
|
"""Resolve a '/Sim/players'-style path against labelled nodes."""
|
|
cur = self.root
|
|
for seg in tagpath.strip("/").split("/"):
|
|
nxt = None
|
|
for c in cur.children:
|
|
base = c.label.split("[")[0]
|
|
if base == seg or c.label == seg:
|
|
nxt = c
|
|
break
|
|
if nxt is None:
|
|
return None
|
|
cur = nxt
|
|
return cur
|
|
|
|
def as_dict(self, depth: int = 2) -> dict:
|
|
return {
|
|
"file": self.path,
|
|
"root": self.digest,
|
|
"policy": {"floats": self.policy, "mask": self.mask,
|
|
"digest": f"blake2b-{DIGEST_BYTES * 8}",
|
|
"readerFingerprint": READER_FINGERPRINT},
|
|
"coverage": self.coverage,
|
|
"readerIssues": {lvl: self.res.count(lvl) for lvl in ("error", "warn", "info")},
|
|
"maskHits": self.mask_hits,
|
|
"subsystems": [{"path": p, "digest": n.hexd, "leaves": n.leaves,
|
|
"valueBytes": n.value_bytes} for p, n in self.subsystems()],
|
|
"tree": self.root.as_dict(depth),
|
|
}
|
|
|
|
|
|
def checksum_bytes(data: bytes, path: str = "<bytes>", floats: str = "bits",
|
|
mask: str = "none", audit: bool = True,
|
|
strict: bool = False) -> SaveChecksum:
|
|
if floats not in FLOAT_POLICIES:
|
|
raise ValueError(f"floats must be one of {FLOAT_POLICIES}")
|
|
if mask not in MASK_PRESETS:
|
|
raise ValueError(f"mask must be one of {tuple(MASK_PRESETS)}")
|
|
res = sr.read_bytes(data, strict=strict)
|
|
masks = MASK_PRESETS[mask]
|
|
hits: dict = {}
|
|
kids = []
|
|
counts = _name_counts(res.tree.children)
|
|
ordinals: dict = {}
|
|
for c in res.tree.children:
|
|
k = ordinals.get(c.name, 0)
|
|
ordinals[c.name] = k + 1
|
|
lab = _label(c, k, None, unique=counts.get(c.name) == 1)
|
|
kids.append(_build(c, "/" + lab, lab, floats, masks, hits))
|
|
root_dig = _h(b"save", floats.encode("ascii"), mask.encode("ascii"),
|
|
*[k.digest for k in kids])
|
|
root = CkNode("", "<save>", "complex", root_dig, 0, len(res.inflated),
|
|
sum(k.leaves for k in kids), sum(k.value_bytes for k in kids), kids)
|
|
cov = audit_coverage(res.tree, res.inflated) if audit else {"ok": None}
|
|
return SaveChecksum(path, root, cov, res, floats, mask, hits)
|
|
|
|
|
|
def checksum_save(path: str, **kw) -> SaveChecksum:
|
|
with open(path, "rb") as f:
|
|
return checksum_bytes(f.read(), path=path, **kw)
|
|
|
|
|
|
# --- diffing ------------------------------------------------------------------
|
|
|
|
class DiffEntry:
|
|
__slots__ = ("path", "kind", "a", "b", "ulps", "note")
|
|
|
|
def __init__(self, path, kind, a=None, b=None, ulps=None, note=""):
|
|
self.path, self.kind, self.a, self.b = path, kind, a, b
|
|
self.ulps, self.note = ulps, note
|
|
|
|
def __repr__(self):
|
|
if self.kind == "value":
|
|
u = f" [{self.ulps:g} ulp]" if self.ulps is not None else ""
|
|
return f"{self.path}: {self.a!r} -> {self.b!r}{u}{self.note}"
|
|
if self.kind == "list":
|
|
return (f"{self.path}: removed {self.a!r}, added {self.b!r}{self.note}")
|
|
return f"{self.path}: {self.kind}{self.note}"
|
|
|
|
def as_dict(self):
|
|
d = {"path": self.path, "kind": self.kind}
|
|
if self.kind in ("value", "list", "list-reordered"):
|
|
d["a"], d["b"] = self.a, self.b
|
|
if self.ulps is not None:
|
|
d["ulps"] = self.ulps
|
|
if self.note:
|
|
d["note"] = self.note.strip()
|
|
return d
|
|
|
|
|
|
def diff(a: CkNode, b: CkNode, out: list | None = None, limit: int = 200) -> list:
|
|
"""Descend two digest trees, reporting only where they differ."""
|
|
out = [] if out is None else out
|
|
if len(out) >= limit:
|
|
return out
|
|
if a.digest == b.digest:
|
|
return out
|
|
if a.children or b.children:
|
|
an = {c.label: c for c in a.children}
|
|
bn = {c.label: c for c in b.children}
|
|
# positional first: order is part of the state
|
|
if [c.label for c in a.children] != [c.label for c in b.children]:
|
|
for lab in an:
|
|
if lab not in bn:
|
|
out.append(DiffEntry(a.path + "/" + lab, "only-in-A"))
|
|
for lab in bn:
|
|
if lab not in an:
|
|
out.append(DiffEntry(b.path + "/" + lab, "only-in-B"))
|
|
if not (set(an) - set(bn)) and not (set(bn) - set(an)):
|
|
out.append(DiffEntry(a.path, "reordered"))
|
|
for ca in a.children:
|
|
cb = bn.get(ca.label)
|
|
if cb is not None:
|
|
diff(ca, cb, out, limit)
|
|
return out
|
|
if a.kind == "list" and b.kind == "list":
|
|
av, bv = a.values or [], b.values or []
|
|
added = [v for v in bv if v not in av]
|
|
removed = [v for v in av if v not in bv]
|
|
if added or removed:
|
|
out.append(DiffEntry(a.path, "list", removed, added,
|
|
note=f" ({len(av)} -> {len(bv)} entries)"))
|
|
else:
|
|
out.append(DiffEntry(a.path, "list-reordered", av, bv))
|
|
return out
|
|
ulps = None
|
|
if a.kind == "float" and b.kind == "float" and len(a.raw) == 4 == len(b.raw):
|
|
ulps = ulps_apart(a.raw, b.raw)
|
|
note = ""
|
|
if a.kind != b.kind:
|
|
note = f" (kind {a.kind} -> {b.kind})"
|
|
av, bv = a.value, b.value
|
|
if a.kind == "raw" or b.kind == "raw":
|
|
# the reader's raw value is a truncated hex preview; two different blobs
|
|
# can preview identically, so report a digest of the whole blob instead
|
|
av = f"<raw {len(a.raw)} B {_h(a.raw).hex()[:SHORT]}>"
|
|
bv = f"<raw {len(b.raw)} B {_h(b.raw).hex()[:SHORT]}>"
|
|
out.append(DiffEntry(a.path, "value", av, bv, ulps, note))
|
|
return out
|
|
|
|
|
|
# --- chain --------------------------------------------------------------------
|
|
|
|
def build_chain(files: list[str], **kw) -> dict:
|
|
"""Record a chain: one entry per save, in turn order."""
|
|
entries = []
|
|
for f in files:
|
|
ck = checksum_save(f, **kw)
|
|
t = ck.res.typed.get("summary") or {}
|
|
entries.append({
|
|
"name": os.path.basename(f),
|
|
"turn": t.get("Turn"),
|
|
"root": ck.digest,
|
|
"coverage": bool(ck.coverage.get("ok")),
|
|
"subsystems": {p: n.hexd for p, n in ck.subsystems()},
|
|
})
|
|
return {"format": "sots-state-chain/1",
|
|
"policy": {"floats": kw.get("floats", "bits"),
|
|
"mask": kw.get("mask", "none"),
|
|
"digest": f"blake2b-{DIGEST_BYTES * 8}",
|
|
"readerFingerprint": READER_FINGERPRINT},
|
|
"turns": entries}
|
|
|
|
|
|
def verify_chain(chain: dict, files: list[str], **kw) -> tuple[bool, list[str]]:
|
|
msgs, ok = [], True
|
|
pol = chain.get("policy") or {}
|
|
for field, got in (("floats", kw.get("floats", "bits")),
|
|
("mask", kw.get("mask", "none"))):
|
|
if pol.get(field) not in (None, got):
|
|
msgs.append(f"!! chain was recorded with {field}={pol[field]!r}, "
|
|
f"verifying with {got!r} -- roots are not comparable")
|
|
ok = False
|
|
fp = pol.get("readerFingerprint")
|
|
if fp and fp != READER_FINGERPRINT:
|
|
msgs.append(f"!! chain was recorded under save_reader {fp}, this is "
|
|
f"{READER_FINGERPRINT} -- re-record before trusting a DIVERGE")
|
|
rec = chain["turns"]
|
|
if len(rec) != len(files):
|
|
msgs.append(f"chain has {len(rec)} turns, {len(files)} saves given")
|
|
ok = False
|
|
for i, f in enumerate(files):
|
|
if i >= len(rec):
|
|
break
|
|
ck = checksum_save(f, **kw)
|
|
want = rec[i]
|
|
if ck.digest == want["root"]:
|
|
msgs.append(f"turn {want.get('turn')} MATCH {ck.digest[:SHORT]} {os.path.basename(f)}")
|
|
continue
|
|
ok = False
|
|
msgs.append(f"turn {want.get('turn')} DIVERGE {ck.digest[:SHORT]} != {want['root'][:SHORT]}"
|
|
f" {os.path.basename(f)}")
|
|
for p, n in ck.subsystems():
|
|
w = want["subsystems"].get(p)
|
|
if w and w != n.hexd:
|
|
msgs.append(f" subsystem {p}: {w[:SHORT]} -> {n.hexd[:SHORT]}")
|
|
msgs.append(" (re-run with the recorded save to localise to an object)")
|
|
break # first divergent turn is the one that matters
|
|
return ok, msgs
|
|
|
|
|
|
# --- CLI ----------------------------------------------------------------------
|
|
|
|
def _mask_note(ck: "SaveChecksum") -> str:
|
|
"""Always say what a mask actually hit -- a mask nobody audits is a hiding place."""
|
|
if ck.mask == "none":
|
|
return ""
|
|
if not ck.mask_hits:
|
|
return " [mask matched NOTHING -- check the rule paths]"
|
|
return " [masked: " + ", ".join(f"{k}x{v}" for k, v in sorted(ck.mask_hits.items())) + "]"
|
|
|
|
|
|
def _print_tree(n: CkNode, depth: int, out, indent: int = 0, top: bool = True):
|
|
if not top:
|
|
pad = " " * indent
|
|
extra = f" {n.leaves} leaves, {n.value_bytes} B" if n.kind in ("complex", "group") else ""
|
|
val = "" if n.kind in ("complex", "group") else f" = {n.value!r}"
|
|
mk = " [masked]" if n.masked else ""
|
|
out.append(f"{pad}{n.short()} {n.label}{val}{extra}{mk}")
|
|
if indent < depth:
|
|
for c in n.children:
|
|
_print_tree(c, depth, out, indent + 1, top=False)
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(
|
|
description="whole-state diagnostic checksum for SOTS1 saves")
|
|
ap.add_argument("saves", nargs="*", help="one save (report), two (diff), or N with --chain")
|
|
ap.add_argument("--floats", choices=FLOAT_POLICIES, default="bits",
|
|
help="float hashing policy (default: bits = exact IEEE-754)")
|
|
ap.add_argument("--mask", choices=tuple(MASK_PRESETS), default="none",
|
|
help="canonicalisation preset (default: none, so deltas localise)")
|
|
ap.add_argument("--tree", action="store_true", help="print the digest tree")
|
|
ap.add_argument("--depth", type=int, default=2, help="tree depth (default 2)")
|
|
ap.add_argument("--json", action="store_true")
|
|
ap.add_argument("--no-audit", action="store_true",
|
|
help="skip the byte-for-byte coverage proof (faster, weaker)")
|
|
ap.add_argument("--strict", action="store_true", help="reader --strict")
|
|
ap.add_argument("--ulps", type=float, default=None,
|
|
help="in a diff, flag float differences at or below N ULPs")
|
|
ap.add_argument("--limit", type=int, default=200, help="max diff entries")
|
|
ap.add_argument("--chain", metavar="FILE", help="chain JSON to verify against")
|
|
ap.add_argument("--record-chain", metavar="FILE", help="write a chain JSON instead")
|
|
args = ap.parse_args(argv)
|
|
|
|
kw = dict(floats=args.floats, mask=args.mask,
|
|
audit=not args.no_audit, strict=args.strict)
|
|
|
|
if not args.saves:
|
|
ap.error("no saves given")
|
|
|
|
if args.record_chain:
|
|
chain = build_chain(args.saves, **kw)
|
|
with open(args.record_chain, "w") as f:
|
|
json.dump(chain, f, indent=1)
|
|
print(f"recorded {len(chain['turns'])} turns -> {args.record_chain}")
|
|
for t in chain["turns"]:
|
|
cov = "cov-ok" if t["coverage"] else "COV-FAIL"
|
|
print(f" turn {t['turn']} {t['root'][:SHORT]} {cov} {t['name']}")
|
|
return 0 if all(t["coverage"] for t in chain["turns"]) else 1
|
|
|
|
if args.chain:
|
|
with open(args.chain) as f:
|
|
chain = json.load(f)
|
|
ok, msgs = verify_chain(chain, args.saves, **kw)
|
|
print("\n".join(msgs))
|
|
return 0 if ok else 1
|
|
|
|
if len(args.saves) == 1:
|
|
ck = checksum_save(args.saves[0], **kw)
|
|
if args.json:
|
|
print(json.dumps(ck.as_dict(args.depth if args.tree else 2), indent=1))
|
|
return 0 if ck.coverage.get("ok") is not False else 1
|
|
cov = ck.coverage
|
|
covs = ("coverage: PROVED (%d bytes rebuilt == inflated)" % cov["rebuiltBytes"]
|
|
if cov.get("ok") else
|
|
"coverage: SKIPPED" if cov.get("ok") is None else
|
|
"coverage: FAILED at inflated offset 0x%x" % (cov.get("firstDiff") or 0))
|
|
print(f"file: {ck.path}")
|
|
print(f"root: {ck.digest}")
|
|
print(f"policy: floats={ck.policy} mask={ck.mask} "
|
|
f"digest=blake2b-{DIGEST_BYTES * 8} reader={READER_FINGERPRINT}"
|
|
f"{_mask_note(ck)}")
|
|
print(f"{covs}; {ck.root.leaves} leaves, {ck.root.value_bytes} value bytes")
|
|
print(f"reader: {ck.res.count('error')} error, {ck.res.count('warn')} warn")
|
|
if args.tree:
|
|
lines: list[str] = []
|
|
_print_tree(ck.root, args.depth, lines)
|
|
print("\n".join(lines))
|
|
else:
|
|
print("subsystems:")
|
|
for p, n in ck.subsystems():
|
|
print(f" {n.short()} {p:<22} {n.leaves:>6} leaves {n.value_bytes:>8} B")
|
|
return 0 if cov.get("ok") is not False else 1
|
|
|
|
if len(args.saves) != 2:
|
|
ap.error("give one save, two saves, or use --chain / --record-chain")
|
|
|
|
a = checksum_save(args.saves[0], **kw)
|
|
b = checksum_save(args.saves[1], **kw)
|
|
entries = diff(a.root, b.root, limit=args.limit)
|
|
if args.ulps is not None:
|
|
for e in entries:
|
|
if e.kind == "value" and e.ulps is not None and e.ulps <= args.ulps:
|
|
e.note += f" <= {args.ulps:g} ULP"
|
|
if args.json:
|
|
print(json.dumps({"a": a.as_dict(0), "b": b.as_dict(0),
|
|
"identical": a.digest == b.digest,
|
|
"diffs": [e.as_dict() for e in entries]}, indent=1))
|
|
return 0 if a.digest == b.digest else 1
|
|
print(f"A {a.digest} {a.path}")
|
|
print(f"B {b.digest} {b.path}")
|
|
print(f"policy: floats={a.policy} mask={a.mask} reader={READER_FINGERPRINT}"
|
|
f"{_mask_note(a)}")
|
|
for ck, nm in ((a, "A"), (b, "B")):
|
|
if ck.coverage.get("ok") is False:
|
|
print(f"!! {nm}: coverage FAILED at 0x{ck.coverage.get('firstDiff') or 0:x} "
|
|
f"-- the digest does not bound this file")
|
|
if a.digest == b.digest:
|
|
print("IDENTICAL")
|
|
return 0
|
|
print(f"DIVERGED: {len(entries)} leaf difference(s)"
|
|
f"{' (limit reached)' if len(entries) >= args.limit else ''}")
|
|
for e in entries:
|
|
print(f" {e}")
|
|
if args.ulps is not None:
|
|
flt = [e for e in entries if e.kind == "value" and e.ulps is not None]
|
|
near = [e for e in flt if e.ulps <= args.ulps]
|
|
print(f"floats: {len(flt)} differ, {len(near)} within {args.ulps:g} ULP")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|