Merge pull request #5310 from springfall2008/fix/coordinator-warn-once

fix(coordinator): warn only about genuine misfilings, once per record

Author: springfall2008

apps/predbat/coordinator.py

@@ -269,10 +269,22 @@ class Redactor:
     # a false-positive rewrite of unrelated data.
     MIN_SUBSTITUTE = 6
 
-    def __init__(self, salt, log=None):
-        """Hold the installation salt and an optional logger for misfiled values."""
+    def __init__(self, salt, log=None, warned=None):
+        """Hold the installation salt, an optional logger for misfiled values, and the set of
+        warnings already logged - owned by the caller so a warning is logged once per process
+        rather than on every redaction (every debug dump and publish)."""
         self.salt = salt
         self.log = log
+        self.warned = warned if warned is not None else set()
+        # Shape-guard hits, held until the walk is done - see _flush_warnings
+        self._pending_warnings = []
+        # device_id of the record being walked, so a warning is deduplicated per record rather
+        # than per field - see _flush_warnings
+        self._record_device_id = None
+        # Every variant of an account identifier noted from a pseudonym container or an
+        # identity-derived device_id: a guard hit wholly explained by one of these is the
+        # redaction working as designed, not a misfiling, so it is not warned about
+        self._identity_variants = set()
         self.originals = {}
         # Subset of self.originals eligible for substring replacement inside a VALUE (never a
         # dict key - see _substitute_key) - true identifiers: account_ids values, an
@@ -290,7 +302,7 @@ def token(self, value):
         digest = hashlib.sha256((self.salt + str(value)).encode("utf-8")).hexdigest()
         return "#" + digest[:8]
 
-    def _note(self, value, substring=False):
+    def _note(self, value, substring=False, identity=False):
         """Record an original so it can later be swapped for its token wherever it appears.
 
         Builds the FULL variant set (see _numeric_variants/_identifier_variants) before minting
@@ -315,7 +327,8 @@ def _note(self, value, substring=False):
         replacement inside a VALUE (never a dict key - see _substitute_key); otherwise each is
         only ever matched by whole-string equality. Registering a variant here unconditionally,
         even one shorter than MIN_SUBSTITUTE, is safe: nothing is ever matched against it below
-        that floor - see _exact_match/_substitute_text.
+        that floor - see _exact_match/_substitute_text. identity=True marks the value as an account
+        identifier reported where one belongs - see _flush_warnings.
         """
         text = str(value)
         variants = {text} | self._numeric_variants(text)
@@ -327,6 +340,8 @@ def _note(self, value, substring=False):
             self.originals[variant] = token
             if substring:
                 self.substring_ok.add(variant)
+            if identity:
+                self._identity_variants.add(variant)
         return token
 
     def _identifier_variants(self, text):
@@ -498,11 +513,14 @@ def _is_location_key(self, name):
 
     def _guard_scalar(self, container, name, value):
         """Pseudonymise one scalar value if it looks misfiled or sits under a location-named field, logging where it was found."""
-        if not (self._misfiled(value, strict_numeric=(container == "hardware_ids")) or self._is_location_key(name)):
+        strict_numeric = container == "hardware_ids"
+        location = self._is_location_key(name)
+        if not (location or self._misfiled(value, strict_numeric=strict_numeric)):
             return value
-        if self.log:
-            label = container if container == name else "{}.{}".format(container, name)
-            self.log("Warn: Coordinator: {} looks like an identifier in a clear container - pseudonymised".format(label))
+        label = container if container == name else "{}.{}".format(container, name)
+        message = "Warn: Coordinator: {} looks like an identifier in a clear container - pseudonymised".format(label)
+        # A location-named field is flagged by its name, whatever the value, so it is always warned about
+        self._pending_warnings.append((message, None if location else value, strict_numeric, self._record_device_id))
         return self._note(value, substring=True)
 
     def _guard_value(self, container, name, value):
@@ -535,12 +553,66 @@ def _guard_key(self, container, name):
         file, exactly what this guard exists to keep out of the debug dump attached to the same
         public issue. Matches _guard_scalar, which never logs the raw value it pseudonymises either.
         """
-        if not self._misfiled(name, strict_numeric=(container == "hardware_ids")):
+        strict_numeric = container == "hardware_ids"
+        if not self._misfiled(name, strict_numeric=strict_numeric):
             return name
-        if self.log:
-            self.log("Warn: Coordinator: {} key looks like an identifier - pseudonymised".format(container))
+        self._pending_warnings.append(("Warn: Coordinator: {} key looks like an identifier - pseudonymised".format(container), name, strict_numeric, self._record_device_id))
         return self._note(name, substring=True)
 
+    def _flush_warnings(self):
+        """Log the shape-guard hits that are genuinely misfiled values, each at most once per process.
+
+        Runs after the walk, once every account identifier in the document has been noted, so a
+        hit can be judged against all of them whatever order the records were walked in. A value
+        that stops looking like an identifier once those are removed - Octopus's
+        "octopus:{mpan}" device_id, a programme's cross-link to it, a station id echoed as a
+        number - is the redaction working as designed: it is still pseudonymised, but warning
+        about it on every debug dump is noise, not advice. The message itself never holds the
+        value (see _guard_key), so it is deduplicated on the message plus the pseudonym of the
+        record's device_id: a second record misfiling the same field is a new leak and warns
+        too, while the same record on the next dump does not.
+        """
+        if not self._pending_warnings:
+            return
+        identities = [self._whole_token(identity) for identity in sorted(self._identity_variants, key=len, reverse=True)]
+        for message, value, strict_numeric, device_id in self._pending_warnings:
+            if value is not None and self._explained_by_identity(value, identities, strict_numeric):
+                continue
+            key = (message, None if device_id is None else self.token(device_id))
+            if key in self.warned:
+                continue
+            self.warned.add(key)
+            if self.log:
+                self.log(message)
+        self._pending_warnings = []
+
+    def _explained_by_identity(self, value, identities, strict_numeric):
+        """Whether a shape-guard hit is wholly an account identifier reported where one belongs.
+
+        `identities` are whole-token patterns (see _whole_token), so a short identifier never
+        splits a longer number's digit run and hides a genuine misfiling. A number is judged by
+        its integer part, as _misfiled does, and is only explained when that IS an identifier -
+        a numeric echo - since stripping digits out of it would rewrite it into something else.
+        """
+        if isinstance(value, bool):
+            return False
+        if isinstance(value, (int, float)):
+            try:
+                text = str(abs(int(value)))
+            except (ValueError, OverflowError):
+                return False
+            return any(identity.fullmatch(text) for identity in identities)
+        # Replaced by a non-digit, non-separator placeholder, not removed: removing it would join
+        # the digit runs either side of it into one that looks like an identifier
+        remaining = str(value)
+        for identity in identities:
+            remaining = identity.sub("|", remaining)
+        return not self._misfiled(remaining, strict_numeric=strict_numeric)
+
+    def _embeds_account_id(self, record):
+        """Whether a record's device_id carries one of its own account identifiers as a whole token."""
+        return any(self._whole_token(value).search(record["device_id"]) for value in self._pseudonym_values(record) if value)
+
     def _has_pseudonym_container(self, node):
         """Whether a pseudonym container (account_ids) sits anywhere in this record - the record
         itself, or a nested sub-record (a meter's tariff, say) - not only its own top-level keys.
@@ -581,11 +653,14 @@ def _walk(self, node, container=None):
         """
         if isinstance(node, dict):
             if isinstance(node.get("device_id"), str) and self._has_pseudonym_container(node) and not self._serial_derived(node):
-                self._note(node["device_id"], substring=True)
+                self._note(node["device_id"], substring=True, identity=self._embeds_account_id(node))
+            outer_device_id = self._record_device_id
+            if isinstance(node.get("device_id"), str):
+                self._record_device_id = node["device_id"]
             out = {}
             for key, value in node.items():
                 if key in PSEUDONYM_CONTAINERS:
-                    out[key] = {self._guard_key(key, name): self._note(entry, substring=True) for name, entry in value.items()}
+                    out[key] = {self._guard_key(key, name): self._note(entry, substring=True, identity=True) for name, entry in value.items()}
                 elif key in CLEAR_CONTAINERS and isinstance(value, dict):
                     out[key] = {self._guard_key(key, name): self._guard_value(key, name, entry) for name, entry in value.items()}
                 elif key in VOCAB_CONTAINERS:
@@ -594,6 +669,7 @@ def _walk(self, node, container=None):
                     out[key] = [self._guard_scalar(key, "token", entry) for entry in value]
                 else:
                     out[key] = self._walk(value, container=key)
+            self._record_device_id = outer_device_id
             return out
         if isinstance(node, list):
             return [self._walk(entry, container=container) for entry in node]
@@ -697,6 +773,7 @@ def redact(self, catalogue):
         serials = sorted(self._collect_serials(catalogue), key=len, reverse=True)
         self._serial_patterns = [self._whole_token(serial) for serial in serials]
         walked = self._walk(catalogue)
+        self._flush_warnings()
         self._substring_order = sorted(self.substring_ok, key=len, reverse=True)
         substituted = self._substitute(walked)
         if "generated" in catalogue:
@@ -728,6 +805,8 @@ def __init__(self, base):
         # the original startup-barrier assemble() call).
         self.assembled = None
         self.salt = None
+        # Redaction warnings already logged - see Redactor._flush_warnings
+        self.redaction_warned = set()
 
     def report(self, component_name, report):
         """Validate and store one component's discovery report, replacing any previous one."""
@@ -905,7 +984,7 @@ def catalogue(self):
         dicts under a lock briefly held to copy self.reports, so redoing it on every debug dump or
         publish() call - its only two callers - costs milliseconds, not a measurable resource.
         """
-        return Redactor(self.load_salt(), log=self.log).redact(self.assemble())
+        return Redactor(self.load_salt(), log=self.log, warned=self.redaction_warned).redact(self.assemble())
 
     def catalogue_raw(self):
         """The assembled catalogue, unredacted. In-process diagnostics only - never write this anywhere.

apps/predbat/tests/test_coordinator.py

@@ -1480,12 +1480,101 @@ def test_guard_key_does_not_log_the_raw_value():
     messages = []
     redactor = Redactor("test-salt-0001", log=messages.append)
     redactor._guard_key("hardware_ids", "1234567890123")
+    redactor._flush_warnings()
     assert any("hardware_ids" in message for message in messages), "expected a warning naming the container: {}".format(messages)
     assert not any("1234567890123" in message for message in messages), "the raw identifier must never reach the log: {}".format(messages)
     print("PASS: _guard_key logs only the container, never the raw key text it is hiding")
     return 0
 
 
+def test_identity_derived_values_do_not_warn():
+    """Octopus reports each meter as "octopus:{mpan}" with the MPAN in account_ids, and a programme
+    cross-links to it. That is the redaction working as designed - every one of those is still
+    pseudonymised - so it must not log a warning on every debug dump, whichever record is walked first."""
+    base, coordinator = _redacting_coordinator()
+    messages = []
+    coordinator.log = messages.append
+    coordinator.report(
+        "octopus",
+        {
+            "programmes": [{"device_id": "axle:site1", "kind": "vpp", "meter": "octopus:1234567890123"}],
+            "meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123", "station": 61234567890}}],
+        },
+    )
+    coordinator.report("solis", {"inverters": [{"device_id": "solis:x1", "ratings": {"echo": 61234567890}}]})
+    catalogue = coordinator.catalogue()
+    text = str(catalogue)
+    assert "1234567890123" not in text and "61234567890" not in text, text
+    assert not [message for message in messages if "identifier" in message], messages
+    print("PASS: an identifier redacted where it belongs, and its echoes, are not warned about")
+    return 0
+
+
+def test_misfiled_identifier_warns_once_per_process():
+    """A genuine misfiling warns, but once - not again on every later debug dump or publish - and an
+    account identifier alongside it does not hide an unrelated identifier sharing the same value."""
+    base, coordinator = _redacting_coordinator()
+    messages = []
+    coordinator.log = messages.append
+    coordinator.report(
+        "octopus",
+        {"meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123"}, "info": {"note": "1234567890123 and 9876543210987"}}]},
+    )
+    for _ in range(3):
+        catalogue = coordinator.catalogue()
+        assert "9876543210987" not in str(catalogue)
+    warnings = [message for message in messages if "identifier" in message]
+    assert warnings == ["Warn: Coordinator: info.note looks like an identifier in a clear container - pseudonymised"], warnings
+    print("PASS: a genuine misfiling warns exactly once across repeated redactions")
+    return 0
+
+
+def test_short_or_unrelated_identities_do_not_hide_misfiled_values():
+    """Suppression is for a value wholly explained by an account identifier: a short identifier
+    must not split a longer number's digit run, a device_id that embeds no account identifier is
+    not identity-derived, and a numeric identifier is judged by its integer part."""
+    messages = []
+    redactor = Redactor("test-salt-0001", log=messages.append)
+    redactor.redact(
+        {
+            "meters": [
+                {"device_id": "octopus:9876543210987", "direction": "import", "account_ids": {"site": "1"}, "info": {"long": "1213141516171"}},
+                {"device_id": "octopus:5", "direction": "import", "account_ids": {"site": "1"}, "info": {"big": 1213141516171}},
+            ]
+        }
+    )
+    warnings = sorted(message for message in messages if "identifier" in message)
+    assert any("info.long" in message for message in warnings), warnings
+    assert any("info.big" in message for message in warnings), warnings
+    assert any("device_id" in message for message in warnings), warnings
+    print("PASS: short or unrelated identities do not hide genuinely misfiled values")
+    return 0
+
+
+def test_warning_is_once_per_record_and_ignores_joined_digit_runs():
+    """A second record misfiling the same field is a new leak and warns too (once per record, not
+    once per field); and removing an explained identifier must not join the digit runs either side
+    of it into one that looks like an identifier."""
+    base, coordinator = _redacting_coordinator()
+    messages = []
+    coordinator.log = messages.append
+
+    def meter(device_id, mpan, note):
+        """One Octopus-shaped meter record with a note in its clear info container."""
+        return {"device_id": device_id, "direction": "import", "account_ids": {"mpan": mpan}, "info": {"note": note}}
+
+    coordinator.report(
+        "octopus",
+        {"meters": [meter("octopus:1111111111111", "1111111111111", "9876543210987"), meter("octopus:2222222222222", "2222222222222", "9876543210988"), meter("octopus:3333333333333", "3333333333333", "ref 12345 3333333333333 67890")]},
+    )
+    for _ in range(2):
+        coordinator.catalogue()
+    warnings = [message for message in messages if "info.note" in message]
+    assert len(warnings) == 2, warnings
+    print("PASS: a warning is once per record, and an explained identifier does not join digit runs")
+    return 0
+
+
 # --- Review round 3: Task 7 review - a case/separator-transformed echo of a pseudonymised value ---
 
 
@@ -2011,6 +2100,10 @@ def test_coordinator_all(my_predbat=None):
     failures += test_shorter_original_does_not_fragment_a_longer_one()
     failures += test_misfiled_identifier_used_as_a_container_key_caught()
     failures += test_guard_key_does_not_log_the_raw_value()
+    failures += test_identity_derived_values_do_not_warn()
+    failures += test_misfiled_identifier_warns_once_per_process()
+    failures += test_short_or_unrelated_identities_do_not_hide_misfiled_values()
+    failures += test_warning_is_once_per_record_and_ignores_joined_digit_runs()
     failures += test_identifier_variants_registered_for_case_and_separator_transforms()
     failures += test_identifier_variants_fold_up_as_well_as_down()
     failures += test_pseudonymised_value_hidden_when_case_folded_and_separator_swapped_in_entity_id()