From b6406a90a4b80671a16b9aa9729a3053ac70b761 Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:27:44 +1100 Subject: [PATCH] Log scrubbing: find names and ssh operands in the original text (#81) Follow-up to #79. Known names and the hosts ssh names are now located as spans in the text as it came, overlapping spans merged, and replaced in one pass. A display name containing a scrubber token (" Jane Doe headset") now goes whole, and a name like "to host" can no longer hide ssh's "connect to host X" operand. Tokens on their own are still left as they are. Co-Authored-By: Claude Opus 5.5 --- tests/test_link.py | 20 ++++++++++++ ui/frame_link.py | 79 +++++++++++++++++++++++++++++++++++++--------- 2 files changed, 84 insertions(+), 15 deletions(-) diff --git a/tests/test_link.py b/tests/test_link.py index e389e9e..ad80c2d 100644 --- a/tests/test_link.py +++ b/tests/test_link.py @@ -1177,6 +1177,26 @@ class FailureLog(unittest.TestCase): "Timed out talking to ") self.assertEqual(fl.hide_hosts("@: ", ["user", "host", "ip"]), "@: ") + def test_a_whole_name_can_cross_a_scrubber_token(self): # #81, first case + self.assertEqual(fl.scrub_failure(" Jane Doe headset stopped answering", [" Jane Doe headset"]), + " stopped answering") + self.assertEqual(fl.hide_hosts("seen Jane Doe headset and ", [" Jane Doe headset", "ip"]), + "seen and ") # the standalone token stays + self.assertEqual(fl.hide_hosts(" frame", ["ip> frame"]), "") # never half a token left + + def test_a_name_cant_swallow_sshs_operand_marker(self): # #81, second case + said = "ssh: connect to host bastion-jane port 22: Connection refused" + out = fl.scrub_failure(said, ["to host"]) + self.assertNotIn("bastion-jane", out) + self.assertEqual(out, "ssh: connect port 22: Connection refused") + err = quiet_log(self) + fl.log_failure("ssh", "ssh stopped", said, hosts=["to host", "host bastion"]) + self.assertNotIn("bastion", err.getvalue()) + self.assertNotIn("jane", err.getvalue()) + # a name overlapping the operand merges with it into one + self.assertEqual(fl.scrub_failure("Could not resolve hostname jane-office now", ["jane-office now"]), + "Could not resolve hostname ") + def test_windows_user_names_with_spaces_go_whole(self): err = quiet_log(self) with mock.patch.object(fl.frame_telemetry, "_user_names", return_value=set()): diff --git a/ui/frame_link.py b/ui/frame_link.py index d4833c8..1207c90 100644 --- a/ui/frame_link.py +++ b/ui/frame_link.py @@ -203,17 +203,58 @@ def probes_summary(probes): PLACEHOLDER = re.compile(r"(<(?:host|user|email|ip|mac|steamid|hex|token|ssh-key|pem|url|redacted)>)") +def _host_spans(text, hosts, keep=()): + """Where each of `hosts` is in text, longest first and whatever the case: (start, end) pairs. + + A name may cross a scrubber's (the whole token then goes with it), but one wholly + inside a token is the token's own word and is left. A name next to one already found sees + it as , as if that one had been replaced first.""" + keep = {k.lower() for k in keep if k} | KEYWORDS # a headset called "host" mustn't eat ssh's wording + names = sorted({h for h in hosts if h and h.lower() not in keep}, key=len, reverse=True) + tokens = [m.span() for m in PLACEHOLDER.finditer(text)] + spans = [] + + def at(i, inside): # the character at i, with names already found standing in as + if not 0 <= i < len(text): + return "" + return inside if any(a <= i < b for a, b in spans) else text[i] + + for h in names: + pat = re.compile(re.escape(h), re.I) + m = pat.search(text) + while m: + s, e = m.span() + before, after, then = at(s - 1, ">"), at(e, "<"), at(e + 1, "<") + ok = (not re.match(r"[\w.:-]", before) and not re.match(r"[\w-]", after) + and not (after in (":", ".", "%") and re.match(r"\w", then))) + if ok and any(a <= s and e <= b for a, b in tokens): + ok = False # "ip" in "": a token already there stays as it is + if ok: + for a, b in tokens: # never cut a token in half: the whole of it goes + if a < e and s < b: + s, e = min(s, a), max(e, b) + spans.append((s, e)) + m = pat.search(text, m.end() if ok else m.start() + 1) + return spans + + +def _hide_spans(text, spans): + """text with each span replaced by , overlapping ones merged into one, in one pass.""" + out, last = [], 0 + for s, e in sorted(spans): + if s < last: # overlaps the one before: that covers this too + last = max(last, e) + continue + out += [text[last:s], ""] + last = e + return "".join(out) + text[last:] + + def hide_hosts(text, hosts, keep=()): """text with each of `hosts` (the headset's own addresses and names) replaced by , longest first and whatever the case, so a bare name like "steamdeck" that the scrubber can't recognise goes too.""" - keep = {k.lower() for k in keep if k} | KEYWORDS # a headset called "host" mustn't eat ssh's wording - names = sorted({h for h in hosts if h and h.lower() not in keep}, key=len, reverse=True) - parts = PLACEHOLDER.split(text) # never inside a scrubber's own already there - for i in range(0, len(parts), 2): - for h in names: - parts[i] = re.sub(r"(?", parts[i], flags=re.I) - return "".join(parts) + return _hide_spans(text, _host_spans(text, hosts, keep)) # ssh names the host it was going to after these words ("Could not resolve hostname X", @@ -223,19 +264,27 @@ PLAIN_WORDS = {"answer", "the", "a", "an", "this", "it", "its", "be", "connect", KEYWORDS = {"host", "hostname", "to", "port"} | PLAIN_WORDS -def hide_operands(text): - def one(m): +def _operand_spans(text): + """Where the hosts ssh names are in text, ssh's own words and any trailing dots left out.""" + spans = [] + for m in OPERAND.finditer(text): word = m.group(2).rstrip(".") - dots = m.group(2)[len(word):] - return m.group(0) if word.lower() in PLAIN_WORDS else f"{m.group(1)} {dots}" - return OPERAND.sub(one, text) + if word.lower() not in PLAIN_WORDS: + spans.append((m.start(2), m.start(2) + len(word))) + return spans + + +def hide_operands(text): + return _hide_spans(text, _operand_spans(text)) def scrub_failure(text, hosts=()): """Free text about a failed attempt, for the log: the attempt's own names, whole and - longest first (case-insensitively, never ssh's own words), then anything ssh names as a - host, then the shared scrubber (addresses, paths, user names).""" - return frame_telemetry.scrub(hide_operands(hide_hosts(str(text or ""), hosts)), 600) + longest first (case-insensitively, never ssh's own words), and anything ssh names as a + host, both found in the text as it came so neither can hide the other, then the shared + scrubber (addresses, paths, user names).""" + text = str(text or "") + return frame_telemetry.scrub(_hide_spans(text, _host_spans(text, hosts) + _operand_spans(text)), 600) def failure_category(message, raw=""):