#!/usr/bin/env python3 """SAFETY-38 round 3 - the SYMMETRIC-PREDICATE FAMILY, enumerated by machine. WHAT IT LOOKS FOR, in one sentence: a call whose argument ORDER is load-bearing for a MESSAGE and for nothing else - so swapping the arguments cannot change a single decision, is invisible to every behavioural check BY CONSTRUCTION, and still inverts what an operator reads on the line they are woken by. ⚠️ WHY IT IS AN INSTRUMENT AND NOT A LIST. MUTATION.md habit 12: "a hand-written scope is the same defect as a hand-written fixture". SAFETY-38 met this family twice - `SignWatchdog.evaluate()` called at `main.py:1620` (round 2) and at `main.py:1560` (round 3), ONE CALL SITE OVER - and both times the enumeration was written by hand and both times it was short. So the scope is computed. ------------------------------------------------------------------------------ THE TEST IT APPLIES For every function in `app/` and every pair (a, b) of its parameters: rename a -> b and b -> a throughout the body, ERASE every string literal and f-string, CANONICALISE the commutative operators, and compare the two ASTs. If they are EQUAL, nothing the function DECIDES can depend on which of the two values arrived first. Then, if some f-string in that function mentions BOTH names, the order is load-bearing for the MESSAGE ALONE, and the pair is reported. ⚠️ THE SYMMETRY IS USUALLY ONE CALL DEEP, WHICH IS WHY THIS IS A FIXPOINT AND NOT A SINGLE PASS. `SignWatchdog.evaluate()` does not compare its two arguments itself; it hands them to `watchdog.disagrees()`, whose body is `commanded_w * against_w < 0 and abs(commanded_w) > min_w and abs(against_w) > min_w`. An intraprocedural scan sees `disagrees(a, b, ...)` become `disagrees(b, a, ...)`, calls that a difference, and reports NOTHING - which is exactly what the first version of this file did, on a tree where the answer was already known. So symmetric pairs discovered in one round are used to canonicalise CALL ARGUMENTS in the next, and the scan iterates until the set stops growing. CANONICALISATION, and each entry is here because something in `app/` needs it: * `a * b`, `a + b` - operand order sorted (`disagrees`) * `x and y`, `x or y` - term order sorted (`disagrees`' conjunction) * `a == b`, `a != b` - operand order sorted * `min(a, b)`, `max(a, b)` - argument order sorted (`clamp_to_rating`) * a tuple/list/set LITERAL being iterated - element order sorted (`rail_ceiling_w`'s `for bound in (dev_min, dev_max)`) * arguments at a KNOWN SYMMETRIC POSITION PAIR of a called function - sorted (the fixpoint) Equal ASTs after the rename is a SUFFICIENT condition for behavioural invariance, not a necessary one. This scan therefore UNDER-REPORTS and says so - see "WHAT THIS CANNOT SEE". That is the conservative direction: everything it names is genuinely in the family, and the gap is stated with its shape. Run: `python3 tools/symmetric_message_scan.py [--quiet]` from goodwe_controller/. Exit status is 0 always; it is a census, not a gate. """ from __future__ import annotations import ast import itertools import pathlib import sys APP = pathlib.Path(__file__).resolve().parent.parent / "app" # {function name: {(i, j) positional index pairs that are interchangeable}} _SYM: dict[str, set[tuple[int, int]]] = {} # ⚠️ AND THE SAME THING FOR KEYWORD-ONLY PARAMETERS, WHICH THE FIRST FIXPOINT # DROPPED. `freeze.rail_ceiling_w(max_w, *, dev_min=None, dev_max=None)` folds # its two device bounds with `for bound in (dev_min, dev_max): ceiling = min( # ceiling, abs(bound))` - perfectly symmetric - and `rail_unreachable()` prints # `"a range of {dev_min}...{dev_max} W"`. A scan that only followed POSITIONAL # arguments called that pair asymmetric and hid a genuine member of the family # behind an implementation detail of its own traversal. _SYM_KW: dict[str, set[tuple[str, str]]] = {} class _Canon(ast.NodeTransformer): """Rename one pair, erase every string, and sort what commutes.""" COMMUTATIVE_BINOP = (ast.Mult, ast.Add, ast.BitAnd, ast.BitOr, ast.BitXor) COMMUTATIVE_CMP = (ast.Eq, ast.NotEq) COMMUTATIVE_CALL = ("min", "max") def __init__(self, a: str = "", b: str = ""): self.a, self.b = a, b # -- the rename --------------------------------------------------------- def visit_Name(self, node): self.generic_visit(node) if self.a and node.id == self.a: node.id = self.b elif self.b and node.id == self.b: node.id = self.a return node def visit_arg(self, node): if self.a and node.arg == self.a: node.arg = self.b elif self.b and node.arg == self.b: node.arg = self.a return node # -- the message is exactly what is ALLOWED to differ -------------------- def visit_JoinedStr(self, node): return ast.Constant(value="") def visit_Constant(self, node): if isinstance(node.value, str): return ast.Constant(value="") return node # -- canonicalisation --------------------------------------------------- def visit_BinOp(self, node): self.generic_visit(node) if isinstance(node.op, self.COMMUTATIVE_BINOP): l, r = ast.dump(node.left), ast.dump(node.right) if l > r: node.left, node.right = node.right, node.left return node def visit_BoolOp(self, node): self.generic_visit(node) node.values = sorted(node.values, key=ast.dump) return node def visit_Compare(self, node): self.generic_visit(node) if (len(node.ops) == 1 and isinstance(node.ops[0], self.COMMUTATIVE_CMP)): l, r = ast.dump(node.left), ast.dump(node.comparators[0]) if l > r: node.left, node.comparators[0] = node.comparators[0], node.left return node def visit_For(self, node): self.generic_visit(node) if isinstance(node.iter, (ast.Tuple, ast.List, ast.Set)): node.iter.elts = sorted(node.iter.elts, key=ast.dump) return node def visit_Call(self, node): self.generic_visit(node) name = (node.func.id if isinstance(node.func, ast.Name) else node.func.attr if isinstance(node.func, ast.Attribute) else "") if name in self.COMMUTATIVE_CALL and len(node.args) == 2 \ and not node.keywords: if ast.dump(node.args[0]) > ast.dump(node.args[1]): node.args = [node.args[1], node.args[0]] return node # the fixpoint: sort the arguments at a known symmetric position pair for (i, j) in sorted(_SYM.get(name, ())): if j < len(node.args): if ast.dump(node.args[i]) > ast.dump(node.args[j]): node.args[i], node.args[j] = node.args[j], node.args[i] # ...and at a known symmetric KEYWORD pair for (ka, kb) in sorted(_SYM_KW.get(name, ())): ia = next((k for k, w in enumerate(node.keywords) if w.arg == ka), None) ib = next((k for k, w in enumerate(node.keywords) if w.arg == kb), None) if ia is not None and ib is not None: va, vb = node.keywords[ia].value, node.keywords[ib].value if ast.dump(va) > ast.dump(vb): node.keywords[ia].value, node.keywords[ib].value = vb, va return node def _canon(src: str, a: str = "", b: str = "") -> str: return ast.dump(_Canon(a, b).visit(ast.parse(src))) # ⚠️ THE SINK IS HALF THE CLAIM, AND THE FIRST VERSION OF THIS SCAN OMITTED IT. # Erasing every f-string makes a function whose ONLY job is to FORMAT one look # perfectly symmetric - `store.unit_key(unit, key)` and # `mqtt.unit_state_key(slug, key)` both came back as "the swap changes nothing" # when in fact the string IS the return value: swapping there renames a STORE # KEY and an MQTT TOPIC, which is a behaviour change every ordinary check can # see, and MAINT-05 is the card about exactly that. The family this file is # looking for is narrower - the string goes to an OPERATOR and to nobody else - # so each sentence is classified by where it GOES. _LOG_SINKS = {"debug", "info", "warning", "warn", "error", "exception", "critical", "log", "log_event", "notify", "persistent_notify"} _MSG_ATTRS = {"reason", "message", "msg", "banner", "detail", "note", "explanation", "why", "text"} def _sink_of(fn, target, follow=True): """RETURN | MESSAGE | OTHER - where this f-string node ends up. `follow` is the ONE step of data flow below, and it is off on the recursive call so that two locals that interpolate each other cannot loop forever. """ for stmt in ast.walk(fn): for child in ast.iter_child_nodes(stmt): pass # a parent map is cheaper than repeated walks parent = {} for node in ast.walk(fn): for child in ast.iter_child_nodes(node): parent[id(child)] = node cur = target while id(cur) in parent: up = parent[id(cur)] if isinstance(up, ast.Return): return "RETURN" if isinstance(up, ast.Raise): return "MESSAGE" if isinstance(up, (ast.Assign, ast.AnnAssign, ast.AugAssign)): tgts = (up.targets if isinstance(up, ast.Assign) else [up.target]) for t in tgts: if isinstance(t, ast.Attribute) and t.attr in _MSG_ATTRS: return "MESSAGE" if isinstance(t, ast.Name) and t.id in _MSG_ATTRS: return "MESSAGE" # ⚠️ ONE STEP OF DATA FLOW, AND IT IS NOT OPTIONAL. `main.py`'s # status page builds the device-range clause into a LOCAL - `_cap = # "; ".join(f"... accepts only {inv.dev_min}...{inv.dev_max} W" ...)` # - and only then concatenates it into the row's `text`. Stopping at # the assignment scores that OTHER and drops a member this card # MEASURED as a live survivor (0 bytes, 21/21 green). So: if the # name assigned here is interpolated into some other sentence in the # same function that IS a message, this one is a message too. if follow: for t in tgts: if isinstance(t, ast.Name) and _feeds_a_message(fn, t.id, up): return "MESSAGE" return "OTHER" # ⚠️ ROUND 4. A SENTENCE HANDED TO A CONSTRUCTOR AS `reason=` IS A # MESSAGE, and scoring only ASSIGNMENTS missed every one of them. This # codebase builds most of its operator sentences that way - # `ArbitrageDecision(action=..., reason=f"...")`, # `out.append({"text": f"..."})` - so the sentence never touches a name # this walker was watching. arbitrage.py's two DISCHARGE/CHARGE reasons # were scored OTHER for exactly this and are genuine members. if isinstance(up, ast.keyword) and up.arg in _MSG_ATTRS: return "MESSAGE" if isinstance(up, ast.Dict): for k, v in zip(up.keys, up.values): if (v is cur and isinstance(k, ast.Constant) and k.value in _MSG_ATTRS): return "MESSAGE" if isinstance(up, ast.Call): name = (up.func.attr if isinstance(up.func, ast.Attribute) else up.func.id if isinstance(up.func, ast.Name) else "") if name in _LOG_SINKS: return "MESSAGE" if name == "append": return "MESSAGE" cur = up return "OTHER" def _feeds_a_message(fn, name, origin): """Is `name` interpolated into some OTHER sentence that IS a message?""" for node in ast.walk(fn): if not isinstance(node, ast.JoinedStr) or node is origin: continue used = any(isinstance(x, ast.Name) and x.id == name for v in node.values if isinstance(v, ast.FormattedValue) for x in ast.walk(v.value)) if used and _sink_of(fn, node, follow=False) == "MESSAGE": return True return False def _fstring_pairs(fn, params): """{(a, b): [(lineno, sink), ...]} for every f-string mentioning both.""" out: dict[tuple[str, str], list[tuple[int, str]]] = {} for node in ast.walk(fn): if not isinstance(node, ast.JoinedStr): continue used = set() for v in node.values: if isinstance(v, ast.FormattedValue): for x in ast.walk(v.value): if isinstance(x, ast.Name) and x.id in params: used.add(x.id) if len(used) < 2: continue sink = _sink_of(fn, node) for a, b in itertools.combinations(sorted(used), 2): out.setdefault((a, b), []).append((node.lineno, sink)) return out def _functions(): for path in sorted(APP.glob("*.py")): tree = ast.parse(path.read_text()) for fn in ast.walk(tree): if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): yield path, fn def _positional(fn): return [a.arg for a in fn.args.posonlyargs + fn.args.args if a.arg != "self"] def _all_params(fn): return _positional(fn) + [a.arg for a in fn.args.kwonlyargs] def _body_src(fn) -> str: """The function's BODY alone. ⚠️ NOT `ast.unparse(fn)`. The signature lists the parameters IN ORDER, so renaming a pair inside it reorders `arguments.args` and every function in the repository compares unequal to itself - which is what the first version of this scan did, silently, and it reported an empty family on a tree whose answer was already known. The claim being tested is about what the body DECIDES, and the body is what is compared. """ return ast.unparse(ast.Module(body=list(fn.body), type_ignores=[])) def _symmetric(fn, a, b) -> bool: src = _body_src(fn) return _canon(src) == _canon(src, a, b) def build_fixpoint(max_rounds=6): """Fill `_SYM` until it stops growing. Returns the number of rounds.""" for round_no in range(1, max_rounds + 1): grew = False for _path, fn in _functions(): pos = _positional(fn) kws = [a.arg for a in fn.args.kwonlyargs] if len(pos) + len(kws) < 2: continue found = set() for i, j in itertools.combinations(range(len(pos)), 2): if _symmetric(fn, pos[i], pos[j]): found.add((i, j)) found_kw = set() for ka, kb in itertools.combinations(sorted(kws), 2): if _symmetric(fn, ka, kb): found_kw.add((ka, kb)) if found - _SYM.get(fn.name, set()): _SYM.setdefault(fn.name, set()).update(found) grew = True if found_kw - _SYM_KW.get(fn.name, set()): _SYM_KW.setdefault(fn.name, set()).update(found_kw) grew = True if not grew: return round_no return max_rounds def scan(): rows = [] for path, fn in _functions(): params = _all_params(fn) if len(params) < 2: continue kwonly = {a.arg for a in fn.args.kwonlyargs} for (a, b), hits in sorted(_fstring_pairs(fn, params).items()): rows.append(dict( module=path.name, func=fn.name, line=fn.lineno, a=a, b=b, msg_lines=sorted(set(hits)), sinks={s for _l, s in hits}, symmetric=_symmetric(fn, a, b), positional=not (a in kwonly or b in kwonly))) return rows # ============================================================================= # CENSUS B - THE SENTENCE THE CALLER BUILDS # ============================================================================= # ⚠️ CENSUS A ABOVE ONLY SEES A CALLEE FORMATTING ITS OWN PARAMETERS, AND THAT # WAS GAP 2 OF ITS OWN "WHAT THIS CANNOT SEE" LIST. The other half of the family # never passes through a function boundary at all: an operator sentence built # where it is printed, interpolating two values that are INTERCHANGEABLE BY # SHAPE. `main.py`'s status page says # # f"{inv.setpoint_entity or inv.name} accepts only {inv.dev_min}...{inv.dev_max} W" # # and transposing those two costs ZERO BYTES, changes nothing any check can # observe, and tells the operator a range backwards. No predicate is involved, # so census A's symmetry test has nothing to test. # # THE TEST: inside one MESSAGE-sink f-string, take every pair of interpolated # expressions, blank every attribute NAME and every string CONSTANT, and compare. # If the blanked forms are EQUAL while the originals DIFFER, the two expressions # are the same shape with one leaf changed - `inv.dev_min` / `inv.dev_max`, # `site_dev_bound("dev_min")` / `site_dev_bound("dev_max")` - which is exactly # the pair a transposition swaps. Same direction as census A: it under-reports # (two values of the same KIND spelled differently are missed) rather than # inventing members. class _Blank(ast.NodeTransformer): def visit_Attribute(self, node): self.generic_visit(node) node.attr = "" return node def visit_Constant(self, node): if isinstance(node.value, str): return ast.Constant(value="") return node def _blanked(expr): return ast.dump(_Blank().visit(ast.parse(ast.unparse(expr)))) def _called_name(func): return (func.attr if isinstance(func, ast.Attribute) else func.id if isinstance(func, ast.Name) else "") def _innermost(tree): """{id(node): INNERMOST enclosing function}, or None at module level. ⚠️ NOT `for fn in ast.walk(tree): for node in ast.walk(fn)`, which is what the first version of census B did. That visits every sentence once per ENCLOSING def, so a sentence inside a nested function was classified twice - once against the inner scope and once against the outer - and the two classifications are not obliged to agree. The rows were de-duplicated afterwards, which hid the double visit rather than removing it. """ m = {} def rec(node, fn): for child in ast.iter_child_nodes(node): f = (child if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) else fn) m[id(child)] = f rec(child, f) rec(tree, None) return m def _spec(fv): """The FORMAT SPEC of one interpolation, as text; '' when there is none. ⚠️ IT IS PRINTED BESIDE EVERY PAIR BECAUSE IT IS THE ONE CHEAP TYPE SIGNAL THIS SCAN HAS, and gap 4 of the second list below is exactly about the pairs that do not share one. `{self.requested_w:+.0f}` beside `{self.applied_w:+.0f}` is two floats rendered identically - a transposition there is silent and reads as a fact. `{self.limit_w:.0f}` beside `{self.bound}` is a float beside a string: a transposition there prints `rating max_w W`, which is visibly broken rather than quietly wrong. Same row in this census, very different hazard, and the reader cannot tell them apart without this column. """ if fv.format_spec is None: return "" return "".join(str(v.value) if isinstance(v, ast.Constant) else "{...}" for v in fv.format_spec.values) def _same_shape_pairs(node): """Pairs of interpolations in ONE f-string that are exactly one leaf apart.""" fvs = [v for v in node.values if isinstance(v, ast.FormattedValue)] for x, y in itertools.combinations(fvs, 2): ua, ub = ast.unparse(x.value), ast.unparse(y.value) if ua == ub: continue if _blanked(x.value) != _blanked(y.value): continue yield ua, ub, _spec(x), _spec(y) def _sentences(): """(path, innermost fn, JoinedStr, sink) for every f-string in app/.""" for path in sorted(APP.glob("*.py")): tree = ast.parse(path.read_text()) inner = _innermost(tree) for node in ast.walk(tree): if not isinstance(node, ast.JoinedStr): continue fn = inner.get(id(node)) if fn is None: continue yield path, fn, node, _sink_of(fn, node) def scan_caller(): rows = [] for path, fn, node, sink in _sentences(): # ⚠️ MESSAGE SINKS ONLY, AND THE FIRST VERSION OF THIS CENSUS SAID # "anything but a RETURN". That let in `p1.py`'s # `self.url = f"http://{self.host}:{self.port}…"` - an f-string that is # DATA, used to make a request, where a transposition is an ordinary # behavioural change every check can see. A formatted string is only in # this family when its only consumer is a person. RETURN is not OTHER # though, and treating it as if it were is what census C below fixes. if sink != "MESSAGE": continue for ua, ub, sa, sb in _same_shape_pairs(node): rows.append((path.name, node.lineno, fn.name, ua, ub, sa, sb)) return sorted(set(rows), key=lambda r: (r[0], r[1], r[3])) # ============================================================================= # CENSUS C - THE SENTENCE THE OBJECT RETURNS ABOUT ITSELF # ============================================================================= # ⚠️ SAFETY-38 ROUND 4. CENSUS A AND CENSUS B BETWEEN THEM COVERED ONE # SYNTACTIC SINK AND CALLED IT THE FAMILY. Every member either census reported # was a sentence APPENDED, ASSIGNED to a `reason`-shaped attribute, or passed as # a LOG ARGUMENT. Not one was a sentence that is RETURNED. Census A had a RETURN # bucket, but only for a callee formatting its own PARAMETERS - so a method # formatting its own ATTRIBUTES and handing the string back was invisible to # both, and `describe()` / `explain()` / the reason-builders are precisely where # this codebase keeps its operator sentences. Two live members were sitting in # that hole: # # clamp.py ClampResult.describe() requested_w <-> applied_w # peakguard.py PeakDecision.explain() projected_w <-> ceiling_w # # Worse than the gap: it was in NONE of the five blind spots this file prints # about itself. A tool that asserts a completeness it does not have is the # recurring defect in a new hat, and the list at the bottom now names its sinks. # # THE TEST is census B's - one f-string, two interpolations one leaf apart - run # over the RETURN sink, plus one thing census B never needed: A RETURNED STRING # IS NOT YET A MESSAGE. `store.unit_key()` and `mqtt.unit_state_key()` return # f-strings too, and swapping there renames a store key and an MQTT topic, which # every ordinary check can see (MAINT-05 is the card about that). So the sink of # a returned sentence is RESOLVED THROUGH ITS CALL SITES: the string is in this # family only when some caller in `app/` feeds it to an operator. def _return_consumers(func_name, depth=0, seen=frozenset()): """[(site, sink, verdict)] - where the STRING RETURNED BY `func_name` goes. One hop is not enough: `Resolution.explain()` is returned again by wrappers that are themselves logged, so a call site whose own sink is RETURN is resolved by recursing on the ENCLOSING function's name, bounded by `depth` and by `seen` so mutual recursion cannot loop. ⚠️ THE MATCH IS ON THE BARE METHOD NAME, exactly as `call_sites()` above, and `explain` is the name of at least four unrelated methods in this tree (`PeakDecision`, `UnitPlan`, `Resolution`, the maintenance migration report). So this OVER-collects sites, which pushes a verdict towards MESSAGE. That is the deliberate direction for a census whose measured failure has twice been under-reporting - but it means the verdict is a PROMPT TO READ THE SITES PRINTED BESIDE IT, not a proof. """ if func_name in seen or depth > 2: return [] seen = seen | {func_name} out = [] for path in sorted(APP.glob("*.py")): tree = ast.parse(path.read_text()) inner = _innermost(tree) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue if _called_name(node.func) != func_name: continue fn = inner.get(id(node)) if fn is None: continue sink = _sink_of(fn, node) label = f"{path.name}:{node.lineno} `{ast.unparse(node.func)}`" if sink == "RETURN": sub = _return_consumers(fn.name, depth + 1, seen) if any(v == "MESSAGE" for _l, _s, v in sub): verdict = "MESSAGE" elif not sub: verdict = "UNRESOLVED" else: verdict = "OTHER" sink = f"RETURN by {fn.name}() -> {verdict}" else: verdict = sink out.append((label, sink, verdict)) return sorted(set(out)) def scan_returned(): """Census C rows: a returned sentence, two interpolations one leaf apart.""" rows = [] for path, fn, node, sink in _sentences(): if sink != "RETURN": continue pairs = list(_same_shape_pairs(node)) if not pairs: continue sites = _return_consumers(fn.name) if any(v == "MESSAGE" for _l, _s, v in sites): verdict = "MESSAGE" elif not sites: verdict = "UNRESOLVED - no call site inside app/" else: verdict = "OTHER" for ua, ub, sa, sb in pairs: rows.append(dict(module=path.name, line=node.lineno, func=fn.name, a=ua, b=ub, spec_a=sa, spec_b=sb, verdict=verdict, sites=sites)) return sorted(rows, key=lambda r: (r["module"], r["line"], r["a"])) # ⚠️ THE UNIT OF COVERAGE IS THE CALL SITE, NOT THE FUNCTION - which is the # whole lesson of SAFETY-38 round 3. `SignWatchdog.evaluate()` had its operand # order pinned and the defect still shipped, because the pin was on the call two # lines below the one a one-inverter roster actually reaches. So a family member # is reported WITH EVERY PLACE IT IS CALLED. def call_sites(name): out = [] for path in sorted(APP.glob("*.py")): tree = ast.parse(path.read_text()) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue if _called_name(node.func) != name: continue # ⚠️ THE RECEIVER IS PRINTED, because matching on the bare # attribute name conflates methods: `evaluate` is the name of # SAFETY-02's, SAFETY-18's and SAFETY-31's watchdogs alike, and # a list of four lines with no receiver tells a reader nothing # about which detector they are looking at. `inv.sign_wd` and # `self.freeze_wd` are the disambiguation, and they are the # caller's own words. out.append(f"{path.name}:{node.lineno} `{ast.unparse(node.func)}`") return out def lazy_log_sites(): """Log calls with a %-format template and >=2 substitution ARGUMENTS. ⚠️ THE SIZE OF THE LARGEST HOLE, MEASURED RATHER THAN ASSERTED. The NOT-SCANNED list below used to end "and app/ uses that form", which is a claim about the tree with no number on it - the exact shape this card has spent four rounds arguing against. `_LOG.error("%s ... %s", a, b)` puts both operands in ARGUMENTS, so no f-string census here can ever reach them, and transposing the two costs the same zero bytes. This counts them. """ out = [] for path in sorted(APP.glob("*.py")): tree = ast.parse(path.read_text()) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue if _called_name(node.func) not in _LOG_SINKS or len(node.args) < 3: continue head = node.args[0] tmpl = "" if isinstance(head, ast.Constant) and isinstance(head.value, str): tmpl = head.value elif isinstance(head, ast.BinOp) or isinstance(head, ast.JoinedStr): tmpl = ast.unparse(head) if tmpl.count("%") >= 2: out.append((path.name, node.lineno, f"{path.name}:{node.lineno} " f"`{ast.unparse(node.func)}` with " f"{len(node.args) - 1} args")) return [r[2] for r in sorted(out)] def _pair_line(sa, sb): same = sa == sb return (f" format spec: '{sa}' / '{sb}' -> " + ("SAME - two values rendered alike, so a transposition is " "SILENT and reads as a fact" if same else "DIFFERENT - likely mixed types, so a transposition is more " "likely to look broken than to lie (gap 4 below)")) def main(): quiet = "--quiet" in sys.argv rounds = build_fixpoint() rows = scan() fam = [r for r in rows if r["symmetric"] and "MESSAGE" in r["sinks"] and "RETURN" not in r["sinks"]] valued = [r for r in rows if r["symmetric"] and "RETURN" in r["sinks"]] print("SYMMETRIC-PREDICATE FAMILY - argument order load-bearing for a " "MESSAGE only") print("=" * 78) print(f"fixpoint reached after {rounds} round(s); " f"{sum(len(v) for v in _SYM.values())} interchangeable positional " f"and {sum(len(v) for v in _SYM_KW.values())} keyword pair(s) " f"across {len(set(_SYM) | set(_SYM_KW))} function(s) in app/") print(f"functions formatting >=2 of their own parameters into one " f"sentence: {len(rows)} pair(s)") print(f"of those, pairs whose SWAP LEAVES EVERY DECISION IDENTICAL: " f"{len(fam) + len(valued)}") print(f" ...of which the sentence is RETURNED (so the swap is a VALUE " f"change an ordinary check can see): {len(valued)}") print(f" ...and the sentence goes ONLY TO AN OPERATOR - THE FAMILY: " f"{len(fam)}") print() print("THE FAMILY - a swap here is invisible to every behavioural check " "BY CONSTRUCTION:") for r in sorted(fam, key=lambda r: (r["module"], r["line"])): kind = ("POSITIONAL - a transposition at a call site costs ZERO BYTES" if r["positional"] else "KEYWORD-ONLY at the definition - a transposition has to be " "written out by name, which is not free and not silent") print(f" {r['module']}:{r['line']} {r['func']}() pair " f"({r['a']}, {r['b']})") print(f" sentence at line(s): " f"{', '.join(f'{l} -> {s}' for l, s in r['msg_lines'])}") print(f" {kind}") sites = call_sites(r["func"]) print(f" CALL SITES ({len(sites)}) - each one is a separate pin, " f"and the receiver says which detector it is:") for site in sites or ["(none inside app/)"]: print(f" {site}") print() print("SYMMETRIC, BUT THE SENTENCE IS THE RETURN VALUE - READ THIS BUCKET " "BY HAND.") print("A returned string is sometimes a KEY (a swap renames a store key or " "an MQTT topic,") print("which is behaviour every ordinary check can see - MAINT-05 is the " "card about it)") print("and sometimes a SENTENCE (a swap is message-only after all, and it " "IS this family):") for r in sorted(valued, key=lambda r: (r["module"], r["line"])): print(f" {r['module']}:{r['line']} {r['func']}() pair " f"({r['a']}, {r['b']})") print() print("=" * 78) print("CENSUS B - SENTENCES THE CALLER BUILDS, interpolating two values of " "the SAME SHAPE") print("A transposition inside one of these is ZERO BYTES and no predicate " "is involved at") print("all, so nothing behavioural can ever see it. Same rule: assert the " "ORDERED PAIR.") cb = scan_caller() print(f"candidates: {len(cb)}") for mod, line, fn, ua, ub, sa, sb in cb: print(f" {mod}:{line} in {fn}()") print(f" {ua} <-> {ub}") print(_pair_line(sa, sb)) print() print("=" * 78) print("CENSUS C - SENTENCES AN OBJECT RETURNS ABOUT ITSELF (SAFETY-38 " "round 4)") print("A method formatting its OWN ATTRIBUTES and RETURNING the string was " "in neither") print("census above: census A only reads a callee's PARAMETERS, census B " "only scored a") print("MESSAGE sink. The verdict below is resolved through the CALL SITES " "of the") print("returning method - a returned string is only in this family once " "some caller") print("hands it to an operator (contrast store.unit_key(), whose return " "value is a KEY).") cc = scan_returned() fam_c = [r for r in cc if r["verdict"] == "MESSAGE"] hot = [r for r in fam_c if r["spec_a"] == r["spec_b"]] print(f"returned sentences with a same-shape pair: {len(cc)} pair(s); " f"{len(fam_c)} reach an operator;") print(f"of those, {len(hot)} ALSO share a format spec - two values " f"rendered alike, which is") print("the hazard gap 4 below names: a transposition there is silent and " "reads as a fact.") print() print("*** = reaches an operator AND shares a format spec - THE ACTIONABLE " "SUBSET") print(" ! = reaches an operator, specs differ (a swap is likelier to look " "broken)") print(" = does not reach an operator, or could not be resolved") # grouped by the sentence, so the CONSUMERS list is printed once per # method rather than once per pair - the ungrouped form was 20 identical # lines under each of six pairs and nobody reads that. for key in sorted({(r["module"], r["line"], r["func"]) for r in cc}): mod, line, fname = key group = [r for r in cc if (r["module"], r["line"], r["func"]) == key] v = group[0]["verdict"] print(f" {mod}:{line} {fname}() verdict: {v}") for r in group: mark = ("***" if r["verdict"] == "MESSAGE" and r["spec_a"] == r["spec_b"] else " ! " if r["verdict"] == "MESSAGE" else " ") print(f" {mark} {r['a']} <-> {r['b']}") print(f" {_pair_line(r['spec_a'], r['spec_b']).strip()}") if not quiet: sites = group[0]["sites"] print(f" CONSUMERS ({len(sites)}) - bare-name match, so read " f"the receiver:") for label, sink, _v in sites or [("(none inside app/)", "", "")]: print(f" {label} -> {sink}") print() print("=" * 78) print("WHAT THIS SCANS, AND WHAT IT DOES NOT - stated because round 4's " "blocker was") print("a completeness this file ASSERTED and did not have. The sink " "classifier is a") print("syntactic walk up the parent chain plus ONE step of local data " "flow, and it") print("recognises exactly these sinks and no others:") print(f" SCANNED as a MESSAGE : a call to {sorted(_LOG_SINKS)}") print(" ...or to `append`; a `raise`; an " "assignment to an") print(f" attribute or local named {sorted(_MSG_ATTRS)}") print(" ...or to a local that is then interpolated " "into one of") print(" those (the one data-flow step).") print(" SCANNED as a RETURN : census C - resolved through the returning " "method's") print(" call sites, one hop plus two of recursion.") print(" NOT SCANNED AT ALL, and each of these can hold a member:") lazy = lazy_log_sites() print(" * `%s`-style lazy logging - `_LOG.error('%s ... %s', a, b)`. The " "arguments") print(" are not an f-string, so no census here looks at them, and a " "transposition") print(" of two `%s` arguments is the same zero-byte edit. THIS IS THE " "LARGEST") print(f" REMAINING HOLE, and it is MEASURED, not asserted: " f"{len(lazy)} such call(s)") print(f" in app/, across " f"{len({s.split(':')[0] for s in lazy})} module(s). None of them is " f"scanned by anything") print(" above. Listed here so the size of the hole is a number:") for site in lazy or ["(none)"]: print(f" {site}") print(" * `.format()`, `%`-formatting, and sentences built by `+` or " "`.join()` out of") print(" non-f-string pieces.") print(" * A sentence that leaves the function inside a DICT, LIST or " "dataclass and is") print(" rendered somewhere else - the one data-flow step is local and " "follows only") print(" a bare Name.") print(" * `yield`ed sentences, and sentences stored on an object by any " "attribute") print(f" name outside {sorted(_MSG_ATTRS)}.") print(" * Anything outside `app/`: `sim/`, `tools/`, `web.py`'s " "templates, the") print(" translations JSON, the firmware, the witness.") print() print("WHAT IT CAN SEE BUT GETS WRONG - the shape of each gap, named " "rather than") print("rounded off:") print(" 1. SEMANTIC symmetry no canonicaliser above spells. The list of " "commutative") print(" forms is finite and hand-written; a function symmetric only " "through a form") print(" not on it is REJECTED and never appears. This under-reports.") print(" 2. In CENSUS B and C, two interpolated values of the same KIND " "that are not the") print(" same SHAPE - `a.lo` beside `hi`, or one of them behind a local " "variable.") print(" Only one-leaf-apart expressions are paired.") print(" 3. WHETHER each call site is pinned. The sites are listed above; " "whether a") print(" check asserts the ORDERED PAIR at each one is a fact about " "`test_*.py` and is") print(" recorded in DECISIONS/SAFETY-38-site-standing-pin.md section " "7. Call sites") print(" outside `app/` are not listed either, and every site list here " "matches on the") print(" bare METHOD NAME - three watchdogs are all called `evaluate` " "and at least") print(" four unrelated classes have an `explain()` - so the receiver is " "printed") print(" beside each one and a reader has to pick.") print(" 4. ⚠️ IT CAN ALSO OVER-REPORT, IN ONE EXACT WAY. The sentence is " "ERASED before") print(" census A's comparison, so a swap that would make the FORMATTING " "ITSELF fail") print(" - a `DAY_NAMES[start]` where `start` is 1320 - is scored as " "'changes no") print(" decision'. `_raise_gap`'s (start, weekday) pairs are that. The " "hazard is a") print(" SAME-TYPED pair, which is why every census B and C row now " "prints whether") print(" the two format specs match.") print(" 5. CENSUS C's verdict is a bare-name resolution (gap 3) and " "therefore reads") print(" MESSAGE too readily. Every verdict is printed with the sites " "that produced") print(" it; a MESSAGE verdict is a prompt to read them, not a proof.") if __name__ == "__main__": main()