From f9f4de31c129eb8cf150265e8bfb57a11ee3cd4b Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Fri, 18 Sep 2026 14:25:45 -0400 Subject: [PATCH] Fixed artifact exporter --- forgetest/forgetest/artifact.py | 8 ++++- forgetest/forgetest/redact.py | 41 ++++++++++++++++++--- forgetest/tests/test_redact.py | 64 +++++++++++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 5 deletions(-) create mode 100644 forgetest/tests/test_redact.py diff --git a/forgetest/forgetest/artifact.py b/forgetest/forgetest/artifact.py index c986309..e7d8cda 100644 --- a/forgetest/forgetest/artifact.py +++ b/forgetest/forgetest/artifact.py @@ -5,6 +5,11 @@ """The release artifact (acceptance.json / acceptance.md) and its verification. +Every test record is redacted before it enters the artifact (redact.py), so +the committed file carries no identity of the machine that ran the campaign. +It happens inside build(), ahead of the self-hash, because verify() rejects a +file edited after the fact. The gate reads none of the redacted fields. + The exporter serializes the campaign state with, for every catalog test, the winning result record (same-campaign PASS or inherited PASS, with its origin) and the fingerprint it was recorded under. The gate @@ -20,6 +25,7 @@ from . import VERSION from . import campaign as _campaign from . import manifest as _manifest from .log import now_ts +from .redact import scrub FORMAT = 1 _RESULT_KEYS = ("ts", "campaign", "test", "result", "fingerprint", "manifest_sha", "image", @@ -52,7 +58,7 @@ def build(state, tests, manifest, records, catalog_hash): "fingerprint": st["fingerprint"], "satisfied": st["satisfied"], "inherited": st["status"] == "inherited", - "record": _record(rec) if rec else None, + "record": scrub(_record(rec)) if rec else None, }) tests_out.append(entry) diff --git a/forgetest/forgetest/redact.py b/forgetest/forgetest/redact.py index 337cc53..f445da4 100644 --- a/forgetest/forgetest/redact.py +++ b/forgetest/forgetest/redact.py @@ -35,7 +35,7 @@ REDACT_KEYS = frozenset(( "hostname", "hostname_file", "host", "ssid", "psk", "passphrase", "password", "passwd", "token", "panel_token", "secret", "api_key", "key", - "mac", "macaddr", "wlan0", "eth0", + "mac", "macaddr", "mac_suffix", "wlan0", "eth0", "gf_username", "gf_password", "username", "user", "email", )) @@ -53,7 +53,11 @@ _CLASSES = ( ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}")), ("EMAIL", re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")), ("MAC", re.compile(r"\b(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\b")), - ("IP6", re.compile(r"\b(?:[0-9A-Fa-f]{1,4}:){2,7}[0-9A-Fa-f]{1,4}\b")), + # Three or more colons, or a "::" run: two colons is a clock time, and + # tagging every timestamp would make the evidence unreadable. + ("IP6", re.compile( + r"(?:[0-9A-Fa-f]{1,4}:){3,7}[0-9A-Fa-f]{1,4}" + r"|[0-9A-Fa-f]{0,4}(?::[0-9A-Fa-f]{1,4})*::(?:[0-9A-Fa-f]{1,4}(?::[0-9A-Fa-f]{1,4})*)?")), ("IP", re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")), ("HEX", re.compile(r"\b[0-9A-Fa-f]{32,}\b")), ) @@ -62,8 +66,28 @@ _CLASSES = ( class Redactor: """Stable per-export numbering, one instance per artifact.""" + # A known value shorter than this is too generic to replace blind. + MIN_KNOWN = 4 + def __init__(self): self._n = {} + self._known = {} + + def learn(self, obj, key=None): + """Collect the values held under the layer-1 keys, so they are + replaced wherever else they appear - a hostname in a log line is + the same leak as a hostname in its own field.""" + if key is not None and key in KEEP_KEYS: + return + if isinstance(obj, dict): + for k, v in obj.items(): + self.learn(v, k) + elif isinstance(obj, list): + for v in obj: + self.learn(v) + elif key in REDACT_KEYS and isinstance(obj, str) and len(obj) >= self.MIN_KNOWN: + if obj not in self._known: + self._known[obj] = self._tag("REDACTED", "%s=%s" % (key, obj)) def _tag(self, cls, value): seen = self._n.setdefault(cls, {}) @@ -72,6 +96,9 @@ class Redactor: return "<%s-%d>" % (cls, seen[value]) def text(self, s): + for known in sorted(self._known, key=len, reverse=True): + if known in s: + s = s.replace(known, self._known[known]) for cls, rx in _CLASSES: def sub(m, cls=cls): v = m.group(0) @@ -87,7 +114,11 @@ class Redactor: if key is not None and key in KEEP_KEYS: return v if key is not None and key in REDACT_KEYS and not isinstance(v, (dict, list)): - return None if v is None else self._tag("REDACTED", "%s=%s" % (key, v)) + if v is None: + return None + if isinstance(v, str) and v in self._known: + return self._known[v] + return self._tag("REDACTED", "%s=%s" % (key, v)) if isinstance(v, dict): return {k: self.value(x, k) for k, x in v.items()} if isinstance(v, list): @@ -99,4 +130,6 @@ class Redactor: def scrub(obj): """Redact a value tree with one export's numbering.""" - return Redactor().value(obj) + r = Redactor() + r.learn(obj) + return r.value(obj) diff --git a/forgetest/tests/test_redact.py b/forgetest/tests/test_redact.py new file mode 100644 index 0000000..3dea5a1 --- /dev/null +++ b/forgetest/tests/test_redact.py @@ -0,0 +1,64 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""The release artifact carries no machine identity (redact.py). + +The artifact is committed to a public repository. These hold the exporter +to that, and to not redacting what the gate recomputes. +""" +import json +import unittest + +from forgetest.redact import scrub + + +class RedactTests(unittest.TestCase): + + def test_keyed_identity_is_replaced(self): + out = scrub({"gf_serial": "15232020", "machine_id": "JTY-876", + "hostname": "forgefirm-b00a"}) + self.assertNotIn("15232020", json.dumps(out)) + self.assertNotIn("JTY-876", json.dumps(out)) + self.assertNotIn("forgefirm-b00a", json.dumps(out)) + + def test_a_known_value_is_replaced_in_free_text_too(self): + # A hostname in a log line is the same leak as one in its own field. + out = scrub({"evidence": {"hostname": "forgefirm-b00a"}, + "log": ["set hostname 'forgefirm-b00a' ok"]}) + self.assertNotIn("forgefirm-b00a", json.dumps(out)) + + def test_addresses_and_macs_go(self): + out = scrub(["ping 172.16.1.97", "wlan0 2C:6B:7D:0D:B0:0A up"]) + t = json.dumps(out) + self.assertNotIn("172.16.1.97", t) + self.assertNotIn("2C:6B:7D:0D:B0:0A", t) + + def test_loopback_is_kept(self): + self.assertIn("127.0.0.1", json.dumps(scrub(["curl http://127.0.0.1:8090/"]))) + + def test_numbering_is_stable_within_an_export(self): + out = scrub(["a 10.0.0.5", "b 10.0.0.6", "c 10.0.0.5"]) + self.assertEqual(out[0].split()[1], out[2].split()[1]) + self.assertNotEqual(out[0].split()[1], out[1].split()[1]) + + def test_integrity_values_survive(self): + # The gate recomputes these; a long hex blob is what they look like. + rec = {"fingerprint": "a" * 64, "manifest_sha": "b" * 64, + "source_sha": "c" * 64, "sha256": "d" * 64, + "catalog_hash": "e" * 64, "identity_sha": "f" * 64} + self.assertEqual(scrub(rec), rec) + + def test_a_clock_time_is_not_an_address(self): + out = scrub(["2026-09-18T18:50:11Z baseline: pre: clean"]) + self.assertIn("18:50:11", out[0]) + + def test_result_fields_the_gate_reads_are_untouched(self): + rec = {"result": "PASS", "ts": "2026-09-18T18:50:11Z", + "campaign": "c-1", "fingerprint": "a" * 64} + self.assertEqual(scrub(rec), rec) + + +if __name__ == "__main__": + unittest.main()