diff --git a/messagefoundry/config/codeset_edit.py b/messagefoundry/config/codeset_edit.py index 9a0826b7..a61c3b49 100644 --- a/messagefoundry/config/codeset_edit.py +++ b/messagefoundry/config/codeset_edit.py @@ -40,6 +40,7 @@ load_code_sets, ) from messagefoundry.config.wiring import WiringError +from messagefoundry.controlchars import has_control_char from messagefoundry.spreadsheet import SPREADSHEET_FORMULA_TRIGGERS, spreadsheet_safe #: Extensions the loader recognises (the writer only ever *writes* ``.csv``; both are read). @@ -302,7 +303,7 @@ def _validate_name(codesets_dir: Path, name: str) -> None: # Reject control characters (NUL, tab, newline, DEL, …) before they reach a filesystem call: an # embedded NUL makes Path.resolve() raise a bare ValueError the CLI's except clause can't catch # (crashing with no JSON on stdout), and none belong in a bare file stem regardless. - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in name): + if has_control_char(name): raise WiringError(f"code set name {name!r} must not contain control characters") if "/" in name or "\\" in name: raise WiringError(f"code set name {name!r} must not contain a path separator") diff --git a/messagefoundry/config/impact.py b/messagefoundry/config/impact.py index c11089e1..6293b837 100644 --- a/messagefoundry/config/impact.py +++ b/messagefoundry/config/impact.py @@ -44,6 +44,7 @@ build_reference_index, ) from messagefoundry.config.wiring import Registry, WiringError +from messagefoundry.controlchars import has_control_char __all__ = [ "LiteralEdit", @@ -628,7 +629,7 @@ def _validate_new_name(config_dir: Path, target_kind: str, new: str) -> None: # A name is embedded verbatim into a Python/TOML string literal; a quote, backslash, or control # char would break out of the literal (or the value). Names are simple identifiers-in-practice — # refuse anything that could corrupt the rewrite. - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in new): + if has_control_char(new): raise WiringError(f"the new name {new!r} must not contain control characters") if any(ch in new for ch in ("'", '"', "\\")): raise WiringError(f"the new name {new!r} must not contain a quote or backslash") diff --git a/messagefoundry/controlchars.py b/messagefoundry/controlchars.py new file mode 100644 index 00000000..f4308797 --- /dev/null +++ b/messagefoundry/controlchars.py @@ -0,0 +1,56 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""The C0/DEL control-character test, written once (BACKLOG #1253). + +WHAT THIS REPLACES. ``ord(ch) < 0x20 or ord(ch) == 0x7F`` was written out seven times across six +files -- two in ``transports/fhir.py`` and one each in ``config/codeset_edit.py``, +``config/impact.py``, ``transports/dicomweb.py``, ``transports/remotefile.py`` and +``transports/rest.py``. Every copy agreed, so nothing was mis-screened. The cost was future-tense +and is the one #1239 named: a later hardening applied to one copy silently does not apply to the +rest, and nothing reports the omission. + +THIS SHARES THE PREDICATE, NOT THE ACTION, AND THAT DISTINCTION IS THE DESIGN. #1239 ruled out +"collapsing the call sites into one helper with a flag" because the differing wrappers are +appropriate: a raise suits a path context, a bool suits a filter, and the exceptions differ by layer +(``WiringError`` in config, a PHI-safe negative ACK in FHIR). So each call site keeps its own +refusal and its own message; only the TEST moves here. A flag parameter would have re-created the +coupling the item exists to remove, one indirection further away. + +TWO ACTIONS ARE PRESERVED ON PURPOSE, and one of them must never be "simplified" into the other: + + * REJECT -- six sites. A control character in a value that reaches a URL path, a header, a + filename or a config field is refused outright. + * STRIP -- ``transports/rest.py`` only, on a message-derived header VALUE. That is defensible + rather than a second instance of the mutation pattern the owner ruled against in #1238: that + ruling turns on ``basename()`` converting a path into a valid-but-DIFFERENT target, handing an + attacker a real file. A header value has no such property -- removing CR/LF cannot redirect a + request anywhere -- and ``rest.py`` already REJECTS a header NAME failing its RFC 7230 token + check. Name-rejected, value-stripped, which is principled. + +DELIBERATELY NOT FOLDED IN. ``parsing/sniff.py`` tests the same code points but is a genuinely +different predicate: it is byte-wise rather than character-wise and subtracts an allowlist, because +a text sniffer must tolerate tab, CR and LF. Folding it in would change its behaviour. + +THE POINT IS THE COPYING PRACTICE, not the seven known lines. If you need this test, import it. +""" + +from __future__ import annotations + + +#: C0 controls (U+0000-U+001F) plus DEL (U+007F). NOT a general "is this printable" test: it is +#: deliberately blind to C1 (U+0080-U+009F) and to Unicode separators, because every call site +#: screens values destined for byte-oriented sinks -- a request line, a header, a path -- where C0 +#: and DEL are the injection alphabet. Widening it is a behaviour change at seven call sites at +#: once, which is exactly the leverage this module exists to provide; make it deliberately. +def has_control_char(text: str) -> bool: + """True if ``text`` contains any C0 control character or DEL.""" + return any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in text) + + +def strip_control_chars(text: str) -> str: + """``text`` with every C0 control and DEL removed. + + The strip arm, used where a value must be neutralised rather than refused. See the module + docstring: this is NOT the general remedy and must not be substituted for a rejection. + """ + return "".join(ch for ch in text if not (ord(ch) < 0x20 or ord(ch) == 0x7F)) diff --git a/messagefoundry/transports/dicomweb.py b/messagefoundry/transports/dicomweb.py index bd7f71b6..5013e448 100644 --- a/messagefoundry/transports/dicomweb.py +++ b/messagefoundry/transports/dicomweb.py @@ -45,6 +45,7 @@ from typing import Any from messagefoundry.config.models import ConnectorType, Destination +from messagefoundry.controlchars import has_control_char from messagefoundry.transports.base import ( DeliveryError, DeliveryResponse, @@ -93,7 +94,7 @@ def _reject_url_control_chars(value: str, field: str) -> None: CR/LF in ``study_uid`` would let it split the request line; urllib would reject it with a bare ``ValueError`` at send. Surface it as a clear construction-time ``ValueError`` (caught at ``check``/dry-run as a ``WiringError``) — PHI-safe (names the field, never the value).""" - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value): + if has_control_char(value): raise ValueError(f"DICOMweb {field} contains an illegal control character") diff --git a/messagefoundry/transports/fhir.py b/messagefoundry/transports/fhir.py index 7293e95c..ac9b0cdd 100644 --- a/messagefoundry/transports/fhir.py +++ b/messagefoundry/transports/fhir.py @@ -47,6 +47,7 @@ from typing import Any from messagefoundry.config.models import ConnectorType, Destination +from messagefoundry.controlchars import has_control_char from messagefoundry.parsing.fhir import FhirPeek, FhirPeekError from messagefoundry.transports.base import ( DeliveryError, @@ -175,7 +176,7 @@ def _reject_control_chars(value: str, field: str) -> str: ``ValueError`` that would otherwise escape ``send()`` as an 'internal error'. Surface it as a permanent ``NegativeAckError`` (a retry re-sends the same body) with a PHI-safe message (the field name only, never the value).""" - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value): + if has_control_char(value): raise NegativeAckError( f"FHIR {field} contains an illegal control character", code="bad-request-value", @@ -198,7 +199,7 @@ def _reject_config_control_chars(value: str, setting: str, where: str = "destina and the read executor -- and #1241's subject is precisely the asymmetry of screening one and not its sibling. """ - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value): + if has_control_char(value): raise ValueError(f"FHIR {where} {setting!r} contains an illegal control character") return value diff --git a/messagefoundry/transports/remotefile.py b/messagefoundry/transports/remotefile.py index d1b2e9cd..ca9ced8b 100644 --- a/messagefoundry/transports/remotefile.py +++ b/messagefoundry/transports/remotefile.py @@ -64,6 +64,7 @@ relax_verify_expiry, resolve_trust_anchor, ) +from messagefoundry.controlchars import has_control_char from messagefoundry.transports.base import ( DeliveryError, DestinationConnector, @@ -118,7 +119,7 @@ def _is_contained_name(name: object) -> bool: return False if "/" in name or "\\" in name: return False - if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in name): + if has_control_char(name): return False # A drive-relative path ("C:x.hl7") resolves against the drive's CWD on Windows and contains no # separator, so the checks above cannot see it. Two chars, ASCII letter, then a colon. diff --git a/messagefoundry/transports/rest.py b/messagefoundry/transports/rest.py index ea1f8dd9..d6fbf7ad 100644 --- a/messagefoundry/transports/rest.py +++ b/messagefoundry/transports/rest.py @@ -54,6 +54,7 @@ is_loopback_hop_host, relax_verify_expiry, ) +from messagefoundry.controlchars import strip_control_chars from messagefoundry.transports.base import ( DeliveryError, DeliveryResponse, @@ -108,7 +109,7 @@ def _strip_header_control_chars(value: str) -> str: """Neutralize a message-derived header VALUE (header-injection safety, #68): strip every C0 control (< 0x20 — incl. CR/LF) and DEL (0x7F) so the value can never split the request line or inject an extra header. Returns the value with those bytes removed (a single, safe header value).""" - return "".join(ch for ch in value if not (ord(ch) < 0x20 or ord(ch) == 0x7F)) + return strip_control_chars(value) # --- captured HTTP response headers (BACKLOG #154, ADR 0013 amendment) ----------------------------- diff --git a/scripts/docs/dangling_citation_check.py b/scripts/docs/dangling_citation_check.py index 067bd888..1aee0144 100644 --- a/scripts/docs/dangling_citation_check.py +++ b/scripts/docs/dangling_citation_check.py @@ -204,7 +204,9 @@ def main(argv: list[str] | None = None) -> int: ) parser.add_argument("paths", nargs="*", type=Path, help="files to scan (default: docs/**/*.md)") parser.add_argument( - "--fail", action="store_true", help="exit non-zero when any citation is unresolved" + "--advisory", + action="store_true", + help="report and exit 0 even when a live-shape citation is found (default: exit 1)", ) args = parser.parse_args(argv) @@ -263,7 +265,24 @@ def main(argv: list[str] | None = None) -> int: print( "Not scanned: the private companion repository, where a citation is invisible to this repo." ) - return 1 if args.fail else 0 + # FAIL CLOSED, ON THE LIVE SHAPE ONLY (BACKLOG #1235). The first version of this took `--fail` + # as opt-in and nothing passed it, so a planted dangling citation was reported correctly AND the + # process still exited 0 -- a checker that cannot fail is not a check, which is the defect this + # tool exists to catch in other people's gates. Flipping the default cost nothing: repo-wide the + # script was referenced by two lines, both inside its own unit test. + # + # The exit code keys on the LIVE shape, not on the hit count. A number at or below the floor can + # never be issued, and a PR/issue/foreign-repo reference is not a backlog citation at all; both + # are reported for a human to read and neither is a defect. Failing on them would red the tree + # today for hits that are correct, and a gate that cries wolf gets switched off. + live = [h for h in hits if h.number > floor and not h.pr_shaped] + if live: + print() + print(f"LIVE SHAPE: {len(live)} citation(s) name a number that can still be issued.") + for h in live: + print(f" {h.path}:{h.lineno}: #{h.number}") + print("Allocate the number before citing it, or write the reference so it CANNOT resolve.") + return 1 if (live and not args.advisory) else 0 if __name__ == "__main__": diff --git a/tests/test_controlchars.py b/tests/test_controlchars.py new file mode 100644 index 00000000..99428e4c --- /dev/null +++ b/tests/test_controlchars.py @@ -0,0 +1,87 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""The shared C0/DEL predicate (BACKLOG #1253). + +#1239 asked for a test that "the two predicates agree across a shared character corpus, so a future +widening of one without the other fails". There is now ONE predicate, so that obligation becomes a +CHARACTERISATION test: pin the exact code-point set, over the whole of Latin-1 plus the neighbours +that tempt a widener, so a change to the shared definition has to be deliberate and cannot ride in +as a tidy-up. Seven call sites move together now -- that is the leverage and also the risk. +""" + +from __future__ import annotations + +import pytest + +from messagefoundry.controlchars import has_control_char, strip_control_chars + +#: The set the predicate is defined to catch. Written independently of the implementation, so this +#: is a second opinion rather than a restatement of the same expression. +_CONTROL = frozenset(chr(c) for c in range(0x00, 0x20)) | {chr(0x7F)} + + +@pytest.mark.parametrize("code", sorted(ord(c) for c in _CONTROL)) +def test_every_c0_control_and_del_is_caught(code: int) -> None: + assert has_control_char(f"a{chr(code)}b") is True + + +def test_the_predicate_matches_its_definition_across_all_of_latin1_and_beyond() -> None: + """The characterisation. Any divergence here is a deliberate widening or a mistake, and either + way it must not pass silently -- seven call sites share this now.""" + caught = {chr(c) for c in range(0x0000, 0x0300) if has_control_char(chr(c))} + assert caught == set(_CONTROL) + + +def test_ordinary_text_is_not_flagged() -> None: + assert has_control_char("") is False + assert has_control_char("a normal value") is False + assert has_control_char("punctuation!@#$%^&*()-_=+[]{};:'\",.<>/?\\|`~") is False + + +def test_the_boundaries_are_where_they_are_documented() -> None: + """0x1F in, 0x20 out; 0x7E out, 0x7F in, 0x80 out. The off-by-one at each edge.""" + assert has_control_char(chr(0x1F)) is True + assert has_control_char(chr(0x20)) is False # space + assert has_control_char(chr(0x7E)) is False # tilde + assert has_control_char(chr(0x7F)) is True # DEL + assert has_control_char(chr(0x80)) is False # C1 starts here and is NOT covered + + +@pytest.mark.parametrize("code", [0x85, 0x9B, 0x2028, 0x2029, 0x200B, 0xFEFF]) +def test_c1_and_unicode_separators_are_deliberately_NOT_caught(code: int) -> None: + """Documented as deliberate, and pinned so nobody "fixes" it by accident. Every call site + screens values bound for byte-oriented sinks where C0 and DEL are the injection alphabet. + Widening this is a behaviour change at seven sites at once and must be made on purpose.""" + assert has_control_char(chr(code)) is False + + +# --- the two actions stay two actions --------------------------------------------------------- + + +def test_strip_removes_exactly_what_the_predicate_catches() -> None: + noisy = "".join(sorted(_CONTROL)) + "keep me" + assert strip_control_chars(noisy) == "keep me" + assert has_control_char(strip_control_chars(noisy)) is False + + +def test_strip_is_a_no_op_on_clean_text() -> None: + assert strip_control_chars("nothing to remove") == "nothing to remove" + + +def test_strip_preserves_order_and_the_rest_of_the_value() -> None: + assert strip_control_chars("a\rb\nc\td") == "abcd" + + +def test_the_two_actions_disagree_on_purpose() -> None: + """A regression that turned the strip into a reject (or vice versa) would show up here. #1253 + requires both arms to survive: six sites refuse, rest.py's header-VALUE path neutralises.""" + hostile = "value\r\nX-Injected: 1" + assert has_control_char(hostile) is True + assert strip_control_chars(hostile) == "valueX-Injected: 1" + assert has_control_char(strip_control_chars(hostile)) is False + + +def test_the_strip_defeats_header_injection_which_is_why_it_exists() -> None: + """CRLF is the whole point: a stripped value can no longer split a request line.""" + assert "\r" not in strip_control_chars("a\rb") + assert "\n" not in strip_control_chars("a\nb") diff --git a/tests/test_dangling_citation_check.py b/tests/test_dangling_citation_check.py index 06022e16..42d00b26 100644 --- a/tests/test_dangling_citation_check.py +++ b/tests/test_dangling_citation_check.py @@ -174,3 +174,53 @@ def test_the_floor_is_conservative_never_optimistic(tmp_path: pathlib.Path) -> N ledger = tmp_path / "L.md" ledger.write_text("## 1500. an item\n\n> open\n", encoding="utf-8") assert cc.allocation_floor([ledger]) == 1500 + + +# --- THE GATE ITSELF, run over the real tree (BACKLOG #1235) --------------------------------------- +# +# The detector shipped in PR #385 wired into NOTHING: repo-wide it was referenced by two lines, both +# inside this file, and its CLI exited 0 even when it reported a hit. A detector nobody invokes and +# that cannot fail is not a gate. These give it the failing arm and make the suite the caller. + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[1] + + +def _live_shape_citations() -> list[tuple[str, int, int]]: + """Citations naming a number ABOVE the floor that are not PR/foreign-repo shaped. + + Below the floor is unreachable forever, and a foreign reference is not a backlog citation at + all; both are reported by the tool for a human to read and neither is a defect. + """ + root = _repo_root() + floor = cc.allocation_floor() + filed = cc.allocated_numbers() + out: list[tuple[str, int, int]] = [] + for path in sorted((root / "docs").rglob("*.md")): + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + for lineno, number, _line, pr_shaped in cc.citations_in(text): + if number in filed or number <= floor or pr_shaped: + continue + out.append((str(path.relative_to(root)), lineno, number)) + return out + + +def test_the_docs_scan_actually_covers_something() -> None: + """PRINT AND PIN THE POPULATION. A walk that collapses to nothing reports clean forever, which + is the exact failure this whole item is about.""" + found = sorted((_repo_root() / "docs").rglob("*.md")) + print(f"scanned {len(found)} markdown files under docs/") + assert len(found) > 200, f"only {len(found)} docs found -- the walk is not finding them" + + +def test_no_docs_citation_names_a_number_that_can_still_be_issued() -> None: + """THE GATE. A citation to an unissued number is harmless until someone files that number, at + which point it silently starts naming unrelated work. Catch it while it is still honest.""" + live = _live_shape_citations() + assert not live, "citations naming a still-issuable number:\n " + "\n ".join( + f"{p}:{n} #{num}" for p, n, num in live + )