diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 5fd541f1..0742078c 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -2916,7 +2916,7 @@ That distinction matters concretely for the ASVS record. The scorecard's absence ## 331. Anonymizer's fail-closed leak-check has no structural PHI detectors -> ๐Ÿ”ข **Filed 2026-08-01 โ€” not started.** Value **6/10** ยท Difficulty **4/10** ยท _quick win_. The function that earns the right to share a de-identified dataset verifies a known-string denylist โ€” `leak_check` is `scan_text` (FORBIDDEN patterns, one routable-IPv4 check, estate substrings; `scripts/security/scan_forbidden.py:772-795`) plus a field-anchored site code, and a real MRN is not a denylisted string โ€” and on a token-less checkout it degrades to the IPv4 check alone over an HL7 body and still returns clean, a gap `f3c6d348` hit in practice with a hand overlay that was never committed; wiring `token_floor_failure()` into the bridge is small, but the unmapped-field report and detectors scoped to fields no rule matched cross the `anonymize` seam and must be mirrored into `tee/anon/leak.py` for `test_anon_parity`. +> โœ… **SHIPPED 2026-08-06 โ€” structural PHI-shape detectors + unmapped-field coverage report + token-floor signal built.** Value **6/10** ยท Difficulty **4/10** ยท _quick win_. `leak_check`/`leak_report` now run high-precision structural detectors (dashed SSN, punctuated NANP phone, CX `MR`/`MRN`-typed identifier) over **the fields no rule matched**, record every present-but-unmapped field in a coverage report (`LeakReport.unmapped_fields`, carried into the `LeakError` on a refusal and exposed via the `on_report` hook), and record `token_floor_failure()` in every report, folding it into the fail-closed decision under the `require_live_denylist` **opt-in** lever โ€” default off, so a token-less CI/OSS/fork load still passes with the structural detectors as the live backstop; a deployment that must refuse on an unloaded denylist sets the lever. The whole structural block is mirrored byte-identical into `tee/anon/leak.py` with a new engine/tee `leak_report` parity test; each detector was falsified. ADR 0030 ยง5/ยง7/Consequences amended (the "deferred" phrasing was stale). The aggressive/broad-shape tier (bare-digit DOB/SSN, name-like runs) stays deferred by owner call โ€” it mass-false-positives on HL7 bodies dense with dates/order-numbers. **Cluster:** Security & Compliance. **Priority:** P2. **Verdict:** build. **Severity:** medium. diff --git a/docs/adr/0030-anonymization-test-harness-tee.md b/docs/adr/0030-anonymization-test-harness-tee.md index d1d76204..013dafee 100644 --- a/docs/adr/0030-anonymization-test-harness-tee.md +++ b/docs/adr/0030-anonymization-test-harness-tee.md @@ -15,6 +15,18 @@ file-replay loader ([`harness/reconcile/compare.py`](../../harness/reconcile/compare.py), [`harness/load/corpus.py`](../../harness/load/corpus.py)); and the publish leak-gate [`scripts/security/scan_forbidden.py`](../../scripts/security/scan_forbidden.py) (`FORBIDDEN`). +- **AMENDED 2026-08-06 (BACKLOG #331):** the leak-check's structural gap that ยง5 / ยง7 / Consequences + below called *scoped out / a deferred improvement* is now **built** โ€” high-precision structural + PHI-shape detectors (dashed SSN, punctuated NANP phone, CX `MR`/`MRN`-typed identifier) over **the + fields no rule matched**, an unmapped-field **coverage report** (`LeakReport.unmapped_fields`; address + only, never a value, carried into the `LeakError` on a refusal and exposed via `on_report`), and the + `token_floor_failure` signal recorded in every report and folded into the fail-closed decision under + the `require_live_denylist` **opt-in** lever โ€” default off, so a token-less CI/OSS/fork load still + passes with the structural detectors as the live backstop; a deployment that must refuse on an + unloaded denylist sets the lever. Mirrored behaviourally-identical into `tee/anon/leak.py`. The + *aggressive/broad* shape tier (bare-digit DOB/SSN, name-like runs) stays deferred โ€” it + mass-false-positives on HL7 bodies (ยง5). This bullet is the single source of the change; the stale + "deferred" phrasings below point back here. - **Decision in one line:** ship a **pure-stdlib, dependency-free `anon` package** that turns real, messy HL7 v2 into structurally-faithful **PHI-free** datasets via a **two-layer rule model โ€” a declarative field-*selection* map (data) over a code registry of pure surrogate *functions* (logic)** โ€” @@ -266,8 +278,11 @@ silently misses a field is worse than none** โ€” though note the limit in the ne PHI the rule map **missed** sails through the fail-closed gate *clean* unless that field happens to contain a denylisted token โ€” a real MRN is not a denylisted string. **Rule-map completeness is therefore the primary control; the leak-check backstops known *strings*, not missed *fields*.** Adding structural detectors -(MRN/SSN/DOB/phone shape, NANP-reserved vs real) to the post-anon check as a *true* field-level backstop is a -candidate improvement, scoped out for this slice with the residual called out (Consequences). +(MRN/SSN/DOB/phone shape, NANP-reserved vs real) to the post-anon check as a *true* field-level backstop was a +candidate improvement โ€” **built in BACKLOG #331** (AMENDED 2026-08-06; see the status banner): the +high-precision set (dashed SSN, punctuated NANP phone, CX `MR`/`MRN`-typed id) scoped to the UNMAPPED fields, +plus an unmapped-field coverage report (`LeakReport.unmapped_fields`) recorded on every pass and surfaced on +a refusal or via the `on_report` hook. The broad-shape tier stays scoped out (below). ### 6. Integration points @@ -294,7 +309,9 @@ candidate improvement, scoped out for this slice with the residual called out (C 0004](0004-payload-agnostic-ingress.md)) so X12 (`parsing/x12/`)/FHIR/raw plug in later โ€” never HL7-parse a non-HL7 body. **Out:** statistical expert-determination de-id, free-text NLP scrubbing (NTE-3/OBX-5 narrative default to **blunt full-redaction**, ยง3 โ€” not entity-level NLP), **per-component (bare-leaf) surrogates** (the -default encoders are field-level `^`-joined, ยง3), structural PHI detectors in the leak-check (ยง5), and any +default encoders are field-level `^`-joined, ยง3), an **aggressive/broad structural PHI search** (bare-digit +DOB/SSN, name-like alpha runs โ€” the mass-false-positive tier; note the *high-precision* unmapped-field +detectors in the leak-check were **built in #331**, AMENDED 2026-08-06, ยง5), and any re-identification/linkage tooling. ### 8. Relationship to the planned de-id framework @@ -340,7 +357,9 @@ AI-assistant `deidentified` data-scope source (PHI.md ยง9 forward-links to it on not MRN/SSN/DOB/name *shapes*, so a field the rule map missed passes the "fail-closed" gate clean unless it contains a denylisted token. **Rule-map completeness is the primary control**; the leak-check is necessary, not sufficient. Free-text (OBX-5/NTE-3) is the **highest-risk residual** โ€” hence its full-redact default - (ยง3). Structural detectors are a deferred improvement (ยง5/ยง7). + (ยง3). **AMENDED 2026-08-06 (#331):** high-precision structural detectors over the UNMAPPED fields plus an + unmapped-field coverage report are now **built** (that deferral is closed; the broad-shape tier stays + deferred, ยง7) โ€” rule-map completeness remains the primary control. - **Vendored-copy drift โ€” bytes *and* behaviour.** Two copies can diverge; the rule/surrogate/token *files* are mitigated by a CI **byte-parity** check (the existing tee discipline) under one authority. But the engine-side `anon/hl7.py` (delegating to `Message`) and the tee's **standalone stdlib re-encoder** are diff --git a/messagefoundry/anon/__init__.py b/messagefoundry/anon/__init__.py index 18ac3a94..a896602d 100644 --- a/messagefoundry/anon/__init__.py +++ b/messagefoundry/anon/__init__.py @@ -15,19 +15,22 @@ Public surface: * :func:`anonymize` โ€” de-identify one HL7 message (raises nothing PHI-bearing). -* :func:`anonymize_checked` โ€” :func:`anonymize` + a **fail-closed** :func:`leak_check`; raises - :class:`LeakError` (token categories only, never the value) if any known partner/site token - survives. This is how you *earn* the right to write a dataset to a shareable location. -* :func:`leak_check` โ€” forbidden-token hits via the publish-guard authority (ADR 0030 ยง5). +* :func:`anonymize_checked` โ€” :func:`anonymize` + a **fail-closed** :func:`leak_report`; raises + :class:`LeakError` (token categories + PHI shapes/addresses only, never a value) if any known + partner/site token survives **or** a structural PHI shape sits in a field no rule mapped. This is + how you *earn* the right to write a dataset to a shareable location. +* :func:`leak_check` / :func:`leak_report` โ€” token hits + structural PHI-shape detection over the + unmapped fields + the unmapped-field coverage report (ADR 0030 ยง5, BACKLOG #331). """ from __future__ import annotations +from collections.abc import Callable from pathlib import Path from .hl7 import anonymize_message from .keying import Keyer -from .leak import LeakCheckUnavailable, leak_check +from .leak import LeakCheckUnavailable, LeakReport, coverage_clause, leak_check, leak_report from .rules import DEFAULT_RULES, AnonError, FieldRule, RuleError, SurrogateKind, load_rules __all__ = [ @@ -37,11 +40,13 @@ "Keyer", "LeakCheckUnavailable", "LeakError", + "LeakReport", "RuleError", "SurrogateKind", "anonymize", "anonymize_checked", "leak_check", + "leak_report", "load_rules", ] @@ -79,18 +84,37 @@ def anonymize_checked( salt: str, overlay: Path | None = None, rules: tuple[FieldRule, ...] | None = None, + require_live_denylist: bool = False, + on_report: Callable[[LeakReport], None] | None = None, ) -> str: - """:func:`anonymize`, then a fail-closed :func:`leak_check`; raise :class:`LeakError` on any hit. + """:func:`anonymize`, then a fail-closed :func:`leak_report`; raise :class:`LeakError` on any hit. Use this whenever the output may be persisted/shared โ€” a silently-missed token is worse than no - anonymization (ADR 0030 ยง5). The raised error names token *categories* only, never the value. + anonymization (ADR 0030 ยง5). The verification is now two-layered (BACKLOG #331): the known-token + denylist **and** high-precision structural PHI-shape detectors over the fields no rule matched, + scoped by the same ``rules`` the anonymizer applied. The raised error names token *categories* and + field *shapes/addresses* only, never a value, and carries a coverage clause (the count + addresses + of the unmapped fields, whether the denylist tables were live) so a refusal is legible. + + ``require_live_denylist`` makes a non-live token source (``token_floor_reason`` set) a refusal + cause in its own right โ€” the strict lever for a deployment that must not de-identify with the + customer denylist unloaded. It defaults **off**: the structural detectors are the live backstop, + and CI/OSS/fork runs legitimately have no token source. ``on_report`` receives the full + :class:`LeakReport` on both the clean and the refusing path (default: no emission). """ - output = anonymize(raw, salt=salt, overlay=overlay, rules=rules) - hits = leak_check(output) - if hits: + effective = rules if rules is not None else load_rules(overlay) + output = anonymize(raw, salt=salt, rules=effective) + report = leak_report(output, rules=effective) + if on_report is not None: + on_report(report) + causes = list(report.hits) + if require_live_denylist and report.token_floor_reason is not None: + causes.append(f"denylist not live: {report.token_floor_reason}") + if causes: raise LeakError( "anonymized output still carries forbidden token(s): " - + "; ".join(sorted(set(hits))) + + "; ".join(sorted(set(causes))) + " โ€” refusing to emit (fail closed). Extend the rule map for the missed field(s)." + + coverage_clause(report) ) return output diff --git a/messagefoundry/anon/leak.py b/messagefoundry/anon/leak.py index df744318..24469f73 100644 --- a/messagefoundry/anon/leak.py +++ b/messagefoundry/anon/leak.py @@ -9,21 +9,39 @@ its importable :func:`scan_text`. The standalone ``tee/anon/leak.py`` vendors the same token data (held identical by the parity test), since the tee cannot reach ``scripts/``. -The check is the **fail-closed backstop**, not the primary control: it catches known *tokens*, not -structural PHI (a missed MRN field with no denylisted string sails through) โ€” rule-map completeness -is the primary control (ADR 0030 ยง5). It is loaded lazily from the source checkout; an installed -wheel without ``scripts/`` raises a clear error (the anonymizer is a dev/migration tool, always run -from a checkout). +The token denylist alone is a **backstop that only catches known *strings***: a real MRN in a field +the rule map never mapped is not a denylisted token, so it would sail through clean. Two structural +controls close that (BACKLOG #331), scoped to the fields ``anonymize`` did **not** rewrite so an +already-pseudonymized field is never re-flagged: + +* an **unmapped-field coverage report** โ€” every present-but-unmapped field is enumerated in the + :class:`LeakReport` (address only, never its value), so the check's reach is legible and a field + nobody thought to map is **recorded** (carried into the :class:`LeakError` on a refusal, and exposed + via ``on_report``) rather than passing unrecorded; and +* **high-precision structural PHI-shape detectors** over those unmapped values (dashed SSN, + punctuated NANP phone, CX ``MR``/``MRN``-typed identifier) โ€” narrow by design to avoid the mass + false-positives a broad digit-run search would produce on HL7 bodies (ADR 0030 ยง5). + +The token-floor signal (``token_floor_failure``) is recorded in every report +(``token_floor_reason``/``token_tables_live``) and folded into the fail-closed decision when +``require_live_denylist`` is set (default off), so a token-less load is legible to any caller that +inspects the report or opts into refusing on it. + +Loaded lazily from the source checkout; an installed wheel without ``scripts/`` raises a clear error +(the anonymizer is a dev/migration tool, always run from a checkout). """ from __future__ import annotations import importlib.util +import re +from dataclasses import dataclass from functools import lru_cache from pathlib import Path from types import ModuleType -from .surrogates import message_has_site_code +from .rules import FieldRule +from .surrogates import Seps, message_has_site_code, read_message_seps class LeakCheckUnavailable(RuntimeError): @@ -47,16 +65,158 @@ def _scanner() -> ModuleType: ) -def leak_check(text: str) -> list[str]: - """Forbidden-token hits in ``text`` (empty list = clean), using the publish-guard's authority. +# --- structural PHI-shape detection over UNMAPPED fields (BACKLOG #331) ---------------------------- +# EVERYTHING from here to the end of this block is held BYTE-IDENTICAL with tee/anon/leak.py (the +# structural walk depends only on read_message_seps, which the parity test pins byte-for-byte). The +# detectors are deliberately high-precision โ€” a broad digit-run search mass-false-positives on HL7 +# bodies dense with dates/order-numbers/set-ids (ADR 0030 ยง5), so the coverage report, not an +# aggressive heuristic, is the catch-all for shapes these cannot safely flag. + +#: A dashed US SSN ``NNN-NN-NNNN`` not embedded in a longer digit run. +_SSN_DASHED: re.Pattern[str] = re.compile(r"(? list[tuple[str, str]]: + """Every ``(address, value)`` in ``text`` whose whole-field ``SEG-i`` address is **not** in + ``mapped_paths`` and whose value is non-empty โ€” the fields the rule map never touched. + + The MSH control header is skipped whole: its field indexing is off-by-one (MSH-N sits at + split-index N-1) and it carries routing/site data the field-anchored site-code pass already + covers, not patient PHI. ``mapped_paths`` is occurrence-agnostic (a rule applies to every + occurrence of its segment), so the address is the bare ``SEG-i``. Returns ``[]`` when the message + has no parseable MSH (there is no field separator to split on). """ - hits = [str(h) for h in _scanner().scan_text(text, include_estate=True)] - if message_has_site_code(text): - hits.append("site-code pattern") + parsed = read_message_seps(text) + if parsed is None: + return [] + _seps, field_sep = parsed + out: list[tuple[str, str]] = [] + for seg in text.replace("\r\n", "\r").replace("\n", "\r").split("\r"): + if not seg: + continue + fields = seg.split(field_sep) + if fields[0].upper() == "MSH": + continue + seg_id = fields[0] + for i in range(1, len(fields)): + value = fields[i] + if not value: + continue + address = f"{seg_id}-{i}" + if address in mapped_paths: + continue + out.append((address, value)) + return out + + +def _has_mrn_typed_identifier(value: str, seps: Seps) -> bool: + """True if any repetition of ``value`` is a CX with a non-empty id (component 1) and a whole + ``MR``/``MRN`` id-type component โ€” an unmapped medical-record number by HL7 structure, far more + precise than a bare digit-run heuristic.""" + for rep in value.split(seps.repetition): + comps = rep.split(seps.component) + if comps[0] and any(comp.upper() in _MRN_TYPES for comp in comps): + return True + return False + + +def _structural_reasons(value: str, seps: Seps) -> list[str]: + """PHI-safe shape labels for one unmapped field value โ€” the SHAPE only, never the value.""" + reasons: list[str] = [] + if _SSN_DASHED.search(value): + reasons.append("unmapped SSN-shaped value") + if _PHONE_DASHED.search(value) or _PHONE_PAREN.search(value): + reasons.append("unmapped phone-shaped value") + if _has_mrn_typed_identifier(value, seps): + reasons.append("unmapped MRN-typed identifier") + return reasons + + +def structural_phi_hits(text: str, mapped_paths: set[str]) -> list[str]: + """Structural PHI-shape hits over the fields no rule matched โ€” reasons name the shape + field + ADDRESS only (e.g. ``"unmapped SSN-shaped value in GT1-16"``), never the offending value, so the + result is safe to raise/log. Empty when the message has no parseable MSH.""" + parsed = read_message_seps(text) + if parsed is None: + return [] + seps, _field_sep = parsed + hits: list[str] = [] + for address, value in unmapped_field_values(text, mapped_paths): + hits.extend(f"{reason} in {address}" for reason in _structural_reasons(value, seps)) return hits + + +@dataclass(frozen=True) +class LeakReport: + """The full result of a leak-check pass โ€” the token hits that decide the fail-closed outcome plus + the coverage context that makes the check's reach legible (all PHI-safe: addresses and reasons, + never a field value). + + * ``hits`` โ€” every leak reason (token/IP/site + structural); non-empty means refuse. + * ``unmapped_fields`` โ€” the addresses present but matched by no rule (the coverage report). + * ``structural_hits`` โ€” the subset of ``hits`` from the structural PHI-shape detectors. + * ``token_tables_live`` โ€” whether the denylist tables loaded from a real token source. + * ``token_floor_reason`` โ€” why the denylist is not trustworthy, or ``None`` if it is. + """ + + hits: list[str] + unmapped_fields: tuple[str, ...] + structural_hits: list[str] + token_tables_live: bool + token_floor_reason: str | None + + +def leak_report(text: str, *, rules: tuple[FieldRule, ...] | None = None) -> LeakReport: + """The full :class:`LeakReport` for ``text`` using the publish-guard's token authority. + + Token hits use the guard's substring + estate-token mode (ADR 0030 ยง5) plus a field-anchored + site-code check. Structural detection engages **only when ``rules`` is supplied** โ€” the address + of every unmapped field is derived from ``{r.path for r in rules}`` and the high-precision + detectors run over those fields. With ``rules is None`` (a direct/legacy call over a bare string) + structural detection is skipped and the result is byte-for-byte the legacy token-only behaviour. + """ + scanner = _scanner() + token_hits = [str(h) for h in scanner.scan_text(text, include_estate=True)] + if message_has_site_code(text): + token_hits.append("site-code pattern") + if rules is None: + unmapped: tuple[str, ...] = () + structural: list[str] = [] + else: + mapped_paths = {r.path for r in rules} + unmapped = tuple(sorted({addr for addr, _ in unmapped_field_values(text, mapped_paths)})) + structural = structural_phi_hits(text, mapped_paths) + token_tables_live = bool(scanner.TOKENS_PRESENT) + token_floor_reason: str | None = scanner.token_floor_failure() + return LeakReport( + hits=token_hits + structural, + unmapped_fields=unmapped, + structural_hits=structural, + token_tables_live=token_tables_live, + token_floor_reason=token_floor_reason, + ) + + +def leak_check(text: str, *, rules: tuple[FieldRule, ...] | None = None) -> list[str]: + """Forbidden-token + structural PHI hits in ``text`` (empty list = clean) โ€” a thin wrapper over + :func:`leak_report`. ``rules`` scopes the structural PHI-shape detectors to the fields no rule + matched; omitting it (a bare-string call) runs the legacy token-only check. + """ + return leak_report(text, rules=rules).hits + + +def coverage_clause(report: LeakReport) -> str: + """A PHI-safe suffix for a fail-closed message naming what the check reached โ€” the count and + ADDRESSES of the unmapped fields (never their values) and whether the denylist tables were live.""" + live = "yes" if report.token_tables_live else "no" + fields = ", ".join(report.unmapped_fields) if report.unmapped_fields else "none" + return ( + f" (checked {len(report.unmapped_fields)} unmapped field(s): {fields}; " + f"denylist tables live: {live})" + ) diff --git a/tee/anon/__init__.py b/tee/anon/__init__.py index 51bd77a4..df8cdb07 100644 --- a/tee/anon/__init__.py +++ b/tee/anon/__init__.py @@ -11,18 +11,21 @@ Public surface (same shape as the engine's): * :func:`anonymize` โ€” de-identify one HL7 message. -* :func:`anonymize_checked` โ€” :func:`anonymize` + a fail-closed :func:`leak_check`; raises - :class:`LeakError` (token categories only) on any surviving token. -* :func:`leak_check` โ€” forbidden-token hits via the vendored token authority. +* :func:`anonymize_checked` โ€” :func:`anonymize` + a fail-closed :func:`leak_report`; raises + :class:`LeakError` (token categories + PHI shapes/addresses only) on any surviving token or a + structural PHI shape in a field no rule mapped. +* :func:`leak_check` / :func:`leak_report` โ€” token hits + structural PHI-shape detection over the + unmapped fields + the unmapped-field coverage report (vendored twin of the engine's; BACKLOG #331). """ from __future__ import annotations +from collections.abc import Callable from pathlib import Path from .hl7 import anonymize_message from .keying import Keyer -from .leak import leak_check +from .leak import LeakReport, coverage_clause, leak_check, leak_report from .rules import DEFAULT_RULES, AnonError, FieldRule, RuleError, SurrogateKind, load_rules __all__ = [ @@ -31,11 +34,13 @@ "FieldRule", "Keyer", "LeakError", + "LeakReport", "RuleError", "SurrogateKind", "anonymize", "anonymize_checked", "leak_check", + "leak_report", "load_rules", ] @@ -67,14 +72,29 @@ def anonymize_checked( salt: str, overlay: Path | None = None, rules: tuple[FieldRule, ...] | None = None, + require_live_denylist: bool = False, + on_report: Callable[[LeakReport], None] | None = None, ) -> str: - """:func:`anonymize`, then a fail-closed :func:`leak_check`; raise :class:`LeakError` on any hit.""" - output = anonymize(raw, salt=salt, overlay=overlay, rules=rules) - hits = leak_check(output) - if hits: + """:func:`anonymize`, then a fail-closed :func:`leak_report`; raise :class:`LeakError` on any hit. + + Two-layered like the engine's (BACKLOG #331): the known-token denylist plus high-precision + structural PHI-shape detectors over the fields no rule matched. ``require_live_denylist`` (default + off) makes a non-live token source a refusal cause; ``on_report`` receives the :class:`LeakReport` + on both paths. The error names token categories and field shapes/addresses only, never a value. + """ + effective = rules if rules is not None else load_rules(overlay) + output = anonymize(raw, salt=salt, rules=effective) + report = leak_report(output, rules=effective) + if on_report is not None: + on_report(report) + causes = list(report.hits) + if require_live_denylist and report.token_floor_reason is not None: + causes.append(f"denylist not live: {report.token_floor_reason}") + if causes: raise LeakError( "anonymized output still carries forbidden token(s): " - + "; ".join(sorted(set(hits))) + + "; ".join(sorted(set(causes))) + " โ€” refusing to emit (fail closed). Extend the rule map for the missed field(s)." + + coverage_clause(report) ) return output diff --git a/tee/anon/leak.py b/tee/anon/leak.py index 0b7c5568..fc041a85 100644 --- a/tee/anon/leak.py +++ b/tee/anon/leak.py @@ -14,17 +14,21 @@ no-op for those (a public checkout has no customer estate to leak). The generic IP detector keeps a literal default so the anonymizer's structural IP check still functions without a token source. -Returns **reasons only** (never the matched text), and is the fail-closed backstop, not the primary -control: it catches known *tokens*, not structural PHI (ADR 0030 ยง5). +Returns **reasons only** (never the matched text), and the token denylist is the fail-closed +*backstop*, not the primary control: it catches known *tokens*, not structural PHI (ADR 0030 ยง5). The +structural PHI-shape detectors + unmapped-field coverage report below (BACKLOG #331) close that gap โ€” +they are held byte-for-byte identical with the engine copy so the two agree on every input. """ from __future__ import annotations import importlib.util import re +from dataclasses import dataclass from pathlib import Path -from .surrogates import message_has_site_code +from .rules import FieldRule +from .surrogates import Seps, message_has_site_code, read_message_seps def _load_publish_guard(_start: Path | None = None) -> object | None: @@ -91,11 +95,165 @@ def scan_text(text: str, *, include_estate: bool = False) -> list[str]: return reasons -def leak_check(text: str) -> list[str]: - """Forbidden-token hits in ``text`` (empty = clean) โ€” the tee's fail-closed leak gate. The site - code is checked **field-anchored** (matching the replace path), so a scrub miss is caught without - false-positiving on a value that merely contains a site-code run.""" - hits = scan_text(text, include_estate=True) - if message_has_site_code(text): - hits.append("site-code pattern") +# --- structural PHI-shape detection over UNMAPPED fields (BACKLOG #331) ---------------------------- +# EVERYTHING from here to the end of this block is held BYTE-IDENTICAL with tee/anon/leak.py (the +# structural walk depends only on read_message_seps, which the parity test pins byte-for-byte). The +# detectors are deliberately high-precision โ€” a broad digit-run search mass-false-positives on HL7 +# bodies dense with dates/order-numbers/set-ids (ADR 0030 ยง5), so the coverage report, not an +# aggressive heuristic, is the catch-all for shapes these cannot safely flag. + +#: A dashed US SSN ``NNN-NN-NNNN`` not embedded in a longer digit run. +_SSN_DASHED: re.Pattern[str] = re.compile(r"(? list[tuple[str, str]]: + """Every ``(address, value)`` in ``text`` whose whole-field ``SEG-i`` address is **not** in + ``mapped_paths`` and whose value is non-empty โ€” the fields the rule map never touched. + + The MSH control header is skipped whole: its field indexing is off-by-one (MSH-N sits at + split-index N-1) and it carries routing/site data the field-anchored site-code pass already + covers, not patient PHI. ``mapped_paths`` is occurrence-agnostic (a rule applies to every + occurrence of its segment), so the address is the bare ``SEG-i``. Returns ``[]`` when the message + has no parseable MSH (there is no field separator to split on). + """ + parsed = read_message_seps(text) + if parsed is None: + return [] + _seps, field_sep = parsed + out: list[tuple[str, str]] = [] + for seg in text.replace("\r\n", "\r").replace("\n", "\r").split("\r"): + if not seg: + continue + fields = seg.split(field_sep) + if fields[0].upper() == "MSH": + continue + seg_id = fields[0] + for i in range(1, len(fields)): + value = fields[i] + if not value: + continue + address = f"{seg_id}-{i}" + if address in mapped_paths: + continue + out.append((address, value)) + return out + + +def _has_mrn_typed_identifier(value: str, seps: Seps) -> bool: + """True if any repetition of ``value`` is a CX with a non-empty id (component 1) and a whole + ``MR``/``MRN`` id-type component โ€” an unmapped medical-record number by HL7 structure, far more + precise than a bare digit-run heuristic.""" + for rep in value.split(seps.repetition): + comps = rep.split(seps.component) + if comps[0] and any(comp.upper() in _MRN_TYPES for comp in comps): + return True + return False + + +def _structural_reasons(value: str, seps: Seps) -> list[str]: + """PHI-safe shape labels for one unmapped field value โ€” the SHAPE only, never the value.""" + reasons: list[str] = [] + if _SSN_DASHED.search(value): + reasons.append("unmapped SSN-shaped value") + if _PHONE_DASHED.search(value) or _PHONE_PAREN.search(value): + reasons.append("unmapped phone-shaped value") + if _has_mrn_typed_identifier(value, seps): + reasons.append("unmapped MRN-typed identifier") + return reasons + + +def structural_phi_hits(text: str, mapped_paths: set[str]) -> list[str]: + """Structural PHI-shape hits over the fields no rule matched โ€” reasons name the shape + field + ADDRESS only (e.g. ``"unmapped SSN-shaped value in GT1-16"``), never the offending value, so the + result is safe to raise/log. Empty when the message has no parseable MSH.""" + parsed = read_message_seps(text) + if parsed is None: + return [] + seps, _field_sep = parsed + hits: list[str] = [] + for address, value in unmapped_field_values(text, mapped_paths): + hits.extend(f"{reason} in {address}" for reason in _structural_reasons(value, seps)) return hits + + +@dataclass(frozen=True) +class LeakReport: + """The full result of a leak-check pass โ€” the token hits that decide the fail-closed outcome plus + the coverage context that makes the check's reach legible (all PHI-safe: addresses and reasons, + never a field value). + + * ``hits`` โ€” every leak reason (token/IP/site + structural); non-empty means refuse. + * ``unmapped_fields`` โ€” the addresses present but matched by no rule (the coverage report). + * ``structural_hits`` โ€” the subset of ``hits`` from the structural PHI-shape detectors. + * ``token_tables_live`` โ€” whether the denylist tables loaded from a real token source. + * ``token_floor_reason`` โ€” why the denylist is not trustworthy, or ``None`` if it is. + """ + + hits: list[str] + unmapped_fields: tuple[str, ...] + structural_hits: list[str] + token_tables_live: bool + token_floor_reason: str | None + + +def leak_report(text: str, *, rules: tuple[FieldRule, ...] | None = None) -> LeakReport: + """The full :class:`LeakReport` for ``text`` using the tee's vendored token authority. + + Behaviourally parallel to the engine's :func:`messagefoundry.anon.leak.leak_report`: the token + hits come from the tee's local :func:`scan_text` (and the field-anchored site-code check) rather + than the engine's ``_scanner()`` delegate, but the structural walk, coverage report, and + token-floor signal are the byte-identical shared logic above. Structural detection engages **only + when ``rules`` is supplied**; a bare-string call is the legacy token-only behaviour. + """ + token_hits = scan_text(text, include_estate=True) + if message_has_site_code(text): + token_hits.append("site-code pattern") + if rules is None: + unmapped: tuple[str, ...] = () + structural: list[str] = [] + else: + mapped_paths = {r.path for r in rules} + unmapped = tuple(sorted({addr for addr, _ in unmapped_field_values(text, mapped_paths)})) + structural = structural_phi_hits(text, mapped_paths) + token_tables_live: bool + token_floor_reason: str | None + if _GUARD is not None: + token_tables_live = bool(_GUARD.TOKENS_PRESENT) # type: ignore[attr-defined] + token_floor_reason = _GUARD.token_floor_failure() # type: ignore[attr-defined] + else: + token_tables_live = False + # nosec B105: a human-readable diagnostic string, not a credential โ€” bandit's + # hardcoded-password heuristic fires only because the name contains "token". + token_floor_reason = "no publish guard reachable โ€” refusing to run structural-only" # nosec B105 + return LeakReport( + hits=token_hits + structural, + unmapped_fields=unmapped, + structural_hits=structural, + token_tables_live=token_tables_live, + token_floor_reason=token_floor_reason, + ) + + +def leak_check(text: str, *, rules: tuple[FieldRule, ...] | None = None) -> list[str]: + """Forbidden-token + structural PHI hits in ``text`` (empty list = clean) โ€” a thin wrapper over + :func:`leak_report`. ``rules`` scopes the structural PHI-shape detectors to the fields no rule + matched; omitting it (a bare-string call) runs the legacy token-only check. + """ + return leak_report(text, rules=rules).hits + + +def coverage_clause(report: LeakReport) -> str: + """A PHI-safe suffix for a fail-closed message naming what the check reached โ€” the count and + ADDRESSES of the unmapped fields (never their values) and whether the denylist tables were live.""" + live = "yes" if report.token_tables_live else "no" + fields = ", ".join(report.unmapped_fields) if report.unmapped_fields else "none" + return ( + f" (checked {len(report.unmapped_fields)} unmapped field(s): {fields}; " + f"denylist tables live: {live})" + ) diff --git a/tests/test_anon_core.py b/tests/test_anon_core.py index 4a3a21e8..dc45e48f 100644 --- a/tests/test_anon_core.py +++ b/tests/test_anon_core.py @@ -22,6 +22,7 @@ anonymize_checked, leak, leak_check, + leak_report, load_rules, ) from messagefoundry.anon.surrogates import Seps, scrub_site_codes, surrogate_field @@ -256,6 +257,139 @@ def test_anonymize_checked_fails_closed_and_is_phi_safe(monkeypatch: pytest.Monk assert "DOE" not in message and "999" not in message # never echoes the body +# --- structural PHI detection on UNMAPPED fields (BACKLOG #331) ------------------------------------ +# The known-token denylist cannot see a real MRN/SSN in a field the rule map never mapped (a real MRN +# is not a denylisted string). These exercise the structural backstop over the UNMAPPED fields. All +# values are SYNTHETIC PHI SHAPES (fake, reserved-fictional, or component-structured) โ€” never a real +# value โ€” and each detector is falsified in the lane report. `DST` is a non-standard segment carrying +# no default rule, so DST-2/3 are the unmapped surface (the f3c6d348 blind-map case in miniature). + +_SSN_MSG = _msg( + r"MSH|^~\&|SAPP|SFAC|RAPP|RFAC|20260101120000||ADT^A01|M1|P|2.5.1", + "PID|1||1^^^H^MR||X^Y", + "DST|1|123-45-6789", # DST-2: unmapped field carrying a synthetic dashed SSN +) + + +@_NO_SCANNER +def test_leak_check_catches_unmapped_ssn() -> None: + """A synthetic dashed SSN in an unmapped field (DST-2) is caught and fails closed. + + Falsified: deleting `_SSN_DASHED` from leak.py's structural set made leak_check() return [] and + anonymize_checked() emit the dataset clean (RED), then restored. + """ + hits = leak_check(_SSN_MSG, rules=DEFAULT_RULES) + assert any("SSN" in h for h in hits), hits + with pytest.raises(LeakError): + anonymize_checked(_SSN_MSG, salt=_SALT) + + +@_NO_SCANNER +def test_leak_check_catches_unmapped_phone() -> None: + """Synthetic punctuated NANP numbers (reserved-fictional 555-01XX) in unmapped fields are caught, + both dashed and parenthesised. + + Falsified: removing the two phone detectors let the dataset slip through clean (RED), then restored. + """ + msg = _msg( + r"MSH|^~\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1", + "PID|1||1^^^H^MR||X^Y", + "DST|1|202-555-0188|(202) 555-0188", # DST-2 dashed, DST-3 parenthesised + ) + hits = leak_check(msg, rules=DEFAULT_RULES) + assert any("phone" in h for h in hits), hits + assert any("DST-2" in h for h in hits) and any("DST-3" in h for h in hits), hits + + +@_NO_SCANNER +def test_leak_check_catches_unmapped_mrn() -> None: + """A CX id typed `MR` in an unmapped field (PID-2, absent from DEFAULT_RULES) is caught by HL7 + structure, and the raw id never surfaces in the reason or the LeakError (PHI-safe). + + Falsified: removing the MR/MRN component detector let the unmapped MRN pass clean (RED), then + restored โ€” confirming the CX id-type signal, not a digit heuristic, is doing the work. + """ + msg = _msg( + r"MSH|^~\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1", + "PID|1|98765^^^HOSP^MR||X^Y", # PID-2: unmapped CX, id-typed MR + ) + hits = leak_check(msg, rules=DEFAULT_RULES) + assert any("MRN" in h and "PID-2" in h for h in hits), hits + assert all("98765" not in h for h in hits) # names the shape + address, never the id + with pytest.raises(LeakError) as exc: + anonymize_checked(msg, salt=_SALT) + assert "98765" not in str(exc.value) + + +@_NO_SCANNER +def test_coverage_report_lists_unmapped_fields() -> None: + """The coverage report enumerates present-but-unmapped fields (address only) โ€” the batch_18 + regression: a field nobody mapped is now visible, not silent. The fail-path LeakError carries the + value-free coverage clause. + + Falsified: stubbing `unmapped_field_values` to yield nothing emptied `.unmapped_fields` (RED), + then restored. + """ + benign = _msg( + r"MSH|^~\&|A|B|C|D|20260101120000||ADT^A01|M1|P|2.5.1", + "PID|1||1^^^H^MR||X^Y", + "DST|1|freeform", # DST-2: unmapped but benign โ€” enumerated, not flagged + ) + report = leak_report(benign, rules=DEFAULT_RULES) + assert "DST-2" in report.unmapped_fields + assert report.structural_hits == [] # benign value โ†’ enumerated only, no shape hit + with pytest.raises(LeakError) as exc: + anonymize_checked(_SSN_MSG, salt=_SALT) + text = str(exc.value) + assert "checked" in text and "unmapped field" in text and "DST-2" in text + + +@_NO_SCANNER +def test_false_positive_guard_benign_unmapped_fields() -> None: + """Unmapped fields dense with dates/coded-values/order-numbers (the mass-false-positive surface + ADR 0030 warns of) must NOT trip the check โ€” why the bare-digit DOB/SSN heuristics were rejected. + + Falsified: broadening `_SSN_DASHED` to any 8+ digit run tripped the 14-digit EVN timestamp (RED), + then restored. + """ + benign = _msg( + r"MSH|^~\&|A|B|C|D|20260101120000||ADT^A01|M1|P|2.5.1", + "EVN|A01|20260101120000", # 14-digit timestamp + "OBX|1|NM|8480-6^Systolic^LN||128|mm[Hg]", # coded observation id + "ORC|NW|1000000042", # unmapped order-number run + "PID|1||1^^^H^MR||X^Y", + ) + assert leak_check(benign, rules=DEFAULT_RULES) == [] + assert anonymize_checked(benign, salt=_SALT) # clean โ†’ returns, no raise + + +@_NO_SCANNER +def test_token_floor_surfaced_when_tables_empty(monkeypatch: pytest.MonkeyPatch) -> None: + """An empty token load is no longer a SILENT green (#331): the report records it and the strict + lever refuses on it โ€” while the default keeps CI/OSS/fork runs (which have no token source) green, + the structural detectors being the live backstop. + + The empty-token state is forced deterministically (this dev checkout has a token source; CI does + not) by patching the loaded scanner's `TOKENS_PRESENT`. Falsified: stubbing `token_floor_failure` + to return None made `.token_floor_reason` None and the strict path stop refusing (RED), restored. + """ + monkeypatch.setattr(leak._scanner(), "TOKENS_PRESENT", False) + clean = _msg( + r"MSH|^~\&|A|B|C|D|20260101120000||ADT^A01|M1|P|2.5.1", + "PID|1||1^^^H^MR||X^Y", + ) + report = leak_report(clean, rules=DEFAULT_RULES) + assert report.token_tables_live is False + assert report.token_floor_reason is not None + # the DEFAULT decision does NOT refuse on empty tokens alone (structural detectors are the backstop) + assert anonymize_checked(clean, salt=_SALT) + # the strict lever DOES refuse, naming the floor reason but no field value + with pytest.raises(LeakError) as exc: + anonymize_checked(clean, salt=_SALT, require_live_denylist=True) + text = str(exc.value) + assert "denylist not live" in text and "fail closed" in text + + def test_alphanumeric_identifier_preserves_width_and_shape() -> None: msg = _msg( r"MSH|^~\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1", diff --git a/tests/test_anon_parity.py b/tests/test_anon_parity.py index 0ddac6e7..d5d05c05 100644 --- a/tests/test_anon_parity.py +++ b/tests/test_anon_parity.py @@ -11,7 +11,9 @@ import pytest +from messagefoundry.anon import DEFAULT_RULES from messagefoundry.anon import anonymize as engine_anonymize +from messagefoundry.anon import leak as engine_leak from messagefoundry.generators import ( _core, _hl7data, @@ -111,6 +113,47 @@ def test_leak_tables_are_sourced_from_the_guard_when_present() -> None: assert tee_leak.FORBIDDEN and tee_leak.ESTATE_TOKENS # type: ignore[attr-defined] +# Synthetic-PHI-shape inputs (never a real value) whose UNMAPPED fields carry SSN/phone/MRN shapes, +# plus a benign case โ€” the structural walk + coverage report + token-floor signal must agree between +# the engine (delegating to _scanner()) and the tee (reimplementing over _GUARD). The structural block +# is copied byte-for-byte between the two leak.py files; this is the divergence guard for it. +_LEAK_PARITY_INPUTS = [ + "MSH|^~\\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1\rPID|1||1^^^H^MR||X^Y\rDST|1|123-45-6789", + "MSH|^~\\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1\rPID|1||1^^^H^MR||X^Y" + "\rDST|1|202-555-0188|(202) 555-0188", + "MSH|^~\\&|A|B|C|D|20260101||ADT^A01|M1|P|2.5.1\rPID|1|98765^^^HOSP^MR||X^Y", + "MSH|^~\\&|A|B|C|D|20260101120000||ADT^A01|M1|P|2.5.1" + "\rEVN|A01|20260101120000\rOBX|1|NM|8480-6^Systolic^LN||128|mm[Hg]", +] + + +def _structural_fields(report: object) -> tuple[object, ...]: + """The STRUCTURAL/coverage fields of a LeakReport โ€” the byte-copied #331 logic this guard pins. + + ``token_tables_live`` / ``token_floor_reason`` are intentionally excluded: they are derived from + the token authority, not the structural walk, and the engine's ``_scanner()`` (lazily lru_cached + on first call) and the tee's ``_GUARD`` (loaded at import) can snapshot the token source at + different times within a full-suite run โ€” a fixture that patches ``MEFOR_FORBIDDEN_TOKENS`` before + the first ``_scanner()`` call poisons its token-floor view for the session. Those fields' cross- + copy agreement is already pinned by ``test_leak_token_table_matches_publish_guard``; here we guard + the detectors + coverage report, which are pure functions of (text, rules). + """ + return (report.hits, report.unmapped_fields, report.structural_hits) # type: ignore[attr-defined] + + +def test_leak_check_and_report_engine_equals_tee() -> None: + for msg in _LEAK_PARITY_INPUTS: + eng_hits = engine_leak.leak_check(msg, rules=DEFAULT_RULES) + tee_hits = tee_leak.leak_check(msg, rules=DEFAULT_RULES) + assert eng_hits == tee_hits, f"leak_check diverged on {msg!r}: {eng_hits!r} != {tee_hits!r}" + eng_report = _structural_fields(engine_leak.leak_report(msg, rules=DEFAULT_RULES)) + tee_report = _structural_fields(tee_leak.leak_report(msg, rules=DEFAULT_RULES)) + assert eng_report == tee_report, ( + f"leak_report structural fields diverged on {msg!r}:" + f"\n ENG {eng_report!r}\n TEE {tee_report!r}" + ) + + def test_adversarial_inputs_engine_output_equals_tee_output() -> None: for msg in _ADVERSARIAL: engine = engine_anonymize(msg, salt=_SALT)