Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion messagefoundry/config/codeset_edit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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")
Expand Down
3 changes: 2 additions & 1 deletion messagefoundry/config/impact.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
build_reference_index,
)
from messagefoundry.config.wiring import Registry, WiringError
from messagefoundry.controlchars import has_control_char

__all__ = [
"LiteralEdit",
Expand Down Expand Up @@ -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")
Expand Down
56 changes: 56 additions & 0 deletions messagefoundry/controlchars.py
Original file line number Diff line number Diff line change
@@ -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))
3 changes: 2 additions & 1 deletion messagefoundry/transports/dicomweb.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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")


Expand Down
5 changes: 3 additions & 2 deletions messagefoundry/transports/fhir.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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

Expand Down
3 changes: 2 additions & 1 deletion messagefoundry/transports/remotefile.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
relax_verify_expiry,
resolve_trust_anchor,
)
from messagefoundry.controlchars import has_control_char
from messagefoundry.transports.base import (
DeliveryError,
DestinationConnector,
Expand Down Expand Up @@ -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.
Expand Down
3 changes: 2 additions & 1 deletion messagefoundry/transports/rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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) -----------------------------
Expand Down
23 changes: 21 additions & 2 deletions scripts/docs/dangling_citation_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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__":
Expand Down
87 changes: 87 additions & 0 deletions tests/test_controlchars.py
Original file line number Diff line number Diff line change
@@ -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")
50 changes: 50 additions & 0 deletions tests/test_dangling_citation_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Loading