diff --git a/changelog.d/2749.security.md b/changelog.d/2749.security.md new file mode 100644 index 0000000000..364d8b3142 --- /dev/null +++ b/changelog.d/2749.security.md @@ -0,0 +1,4 @@ +- **Turning off a size cap on an HTTP, File or SFTP/FTP inbound now logs a warning.** Setting + `max_body_bytes` on an HTTP inbound, or `max_file_bytes` on a File or SFTP/FTP inbound, to `0` or + `None` is still allowed and still removes the cap. The connection now logs a warning each time it + is built, naming the connection and the setting, never the value. (`BACKLOG #2749`) diff --git a/changelog.d/2900.changed.md b/changelog.d/2900.changed.md new file mode 100644 index 0000000000..b932b80375 --- /dev/null +++ b/changelog.d/2900.changed.md @@ -0,0 +1,3 @@ +- **A TCP connection refuses `0x20` (a space) as its `start` or `end` frame byte.** A reply's frame + bytes are replaced by a space, so a space frame byte could not be kept out of a reply. A `0x20` + `trailer` is still allowed. The refusal fires when the connection is built. (`BACKLOG #2900`) diff --git a/changelog.d/3151.changed.md b/changelog.d/3151.changed.md new file mode 100644 index 0000000000..8daef5eb52 --- /dev/null +++ b/changelog.d/3151.changed.md @@ -0,0 +1,3 @@ +- **A File inbound refuses a negative `min_age_seconds` at build.** A negative value used to build + and act as zero. It is now a build error naming the setting, with the value withheld. Zero, the + default, and any positive value build as before. (`BACKLOG #3151`) diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index 7027092eef..a137f75db1 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -482,7 +482,7 @@ the bytes. | `host` | out | — (required) | the downstream peer to dial. **Inbound takes no host** (wiring error) — listeners bind the service-level `[inbound].bind_host`. | | `port` | both | — (required) | bind/connect port | | `framing` | both | `"stx_etx"` | framing **preset**: `"stx_etx"` (`0x02`/`0x03`, no trailer) or `"vt_fs"`/`"mllp"` (`0x0B`/`0x1C`/`0x0D`). Pass `framing=None` to use explicit bytes instead. | -| `start` / `end` / `trailer` | both | — | explicit delimiter **byte ints** (use with `framing=None`; `trailer` optional). Specifying these *and* a preset is a config error. | +| `start` / `end` / `trailer` | both | — | explicit delimiter **byte ints** (use with `framing=None`; `trailer` optional). Specifying these *and* a preset is a config error. `start` and `end` must differ, and neither may be `0x20` (a space); the connection is refused when it is built. A `0x20` `trailer` is allowed. Why: `FrameCodec` in [`messagefoundry/framing.py`](../messagefoundry/framing.py). | | `encoding` | both | `utf-8` | charset used to encode/decode the framed payload | | `max_connections` | in | `256` | cap on concurrent client connections (flood guard). `None`/`0` = unlimited. | | `receive_timeout` | in | `60.0` | close a client idle this many seconds (slowloris). `None`/`0` = no timeout. | @@ -653,7 +653,7 @@ routes a `Message`; `json`/`xml`/`text`/`fhir` route a `RawMessage` the Handler | `max_connections` | `256` | cap on concurrent clients (connection-flood guard). `None`/`0` = unlimited. | | `receive_timeout` | `60.0` | bound the **whole-request** read — request line + headers + body (slowloris guard); over budget answers a synchronous `408`. `None`/`0` = no timeout. | | `max_connections_per_host` | **off** | cap on concurrent connections from **one peer address**, answered `503` with an `at_capacity` event whose reason reads `max_connections_per_host` (vault BACKLOG #2606). **Ships off, unlike MLLP/TCP/X12**: this connector answers one request per connection, so behind a reverse proxy every partner arrives from one address and any cap would be the listener's whole capacity. Set one only where the listener sees real client addresses. | -| `max_body_bytes` | `16 MiB` | the MLLP frame cap's HTTP twin — an over-declared `Content-Length` is refused `413` **before a body byte is read** (OOM guard). `None`/`0` = unlimited. | +| `max_body_bytes` | `16 MiB` | the MLLP frame cap's HTTP twin — an over-declared `Content-Length` is refused `413` **before a body byte is read** (OOM guard). `None`/`0` = unlimited, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `max_header_bytes` | `64 KiB` | cap the request line + headers (header-flood guard). This one cap can't be switched off: unset or `None` takes the 64 KiB default, and `0` (or `"0"`) is refused at load rather than silently becoming the default (BACKLOG #1872). | | `max_messages_per_second` | **off** | sustained message-rate ceiling for the **whole listener** (ASVS 2.4.1 / 15.2.2, BACKLOG #1114 — the MLLP pacer, ported). Over budget the connector **waits before reading the request**, so the partner is back-pressured and then served in full — **nothing is dropped, refused or answered differently**, and the wait sits outside `receive_timeout` so a paced partner is never handed a `408` for a delay the engine imposed. **Listener-wide, not per-connection**, unlike MLLP/TCP/X12: this connector answers one request per connection, so a per-connection bucket would be charged once and thrown away, bounding nothing. A `GET`/`HEAD` probe waits behind an outstanding debt but **charges nothing**, and neither does a request refused before its body reaches the engine. Only a body the engine reads and records spends budget, including one it then refuses with a `422`, so a peer that submits nothing cannot starve one that does. Unset = no bound, deliberately: a guessed rate throttles real traffic, so the number has to come from your own feed profile. | | `message_burst` | = the rate | tokens the bucket holds, i.e. how large a burst passes unpaced before the sustained rate applies. Only meaningful with `max_messages_per_second` set. Floor of 1 so the listener can always make progress. | @@ -779,11 +779,11 @@ def route(msg): | `directory` | both | — (required) | folder to poll / write into | | `pattern` | in | `*.hl7` | filename glob to pick up | | `poll_seconds` | in | `1.0` | poll interval, a finite number of seconds above zero; anything else is refused at build, including at least zero, a negative value, NaN, infinity and text that is not a number (BACKLOG #2774). It is also the **settle window**: a file is read only once its size and modification time are unchanged since the last poll that saw it (BACKLOG #1811), so every file waits at least one poll. The settle gate is always on and has no setting, but a very small `poll_seconds` narrows its window to almost nothing. | -| `min_age_seconds` | in | `0` | skip files modified within this window. This is an extra wait on top of the settle gate, not the gate itself; set it for a partner that pauses between writes for longer than `poll_seconds`. | +| `min_age_seconds` | in | `0` | skip files modified within this window. This is an extra wait on top of the settle gate, not the gate itself; set it for a partner that pauses between writes for longer than `poll_seconds`. A negative value is refused at build (BACKLOG #3151). | | `after_read` | in | `move` | `move` (→ `.processed`), `delete`, or `leave` (process **in place** — never move/delete the source file, for a read-only share / a directory another system owns; a hashed dedup ledger ensures a left file is ingested **once**, #142) | | `sort` | in | `name` | process order: `name` or `mtime` | | `recursive` | in | `false` | also scan subdirectories | -| `max_file_bytes` | in | `16 MiB` | route files larger than this to the error dir instead of reading them into memory (OOM guard). `None`/`0` = unlimited. | +| `max_file_bytes` | in | `16 MiB` | route files larger than this to the error dir instead of reading them into memory (OOM guard). `None`/`0` = unlimited, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `poll_max_files` | in | `500` | most files one scan will take. The rest stay in the drop directory and the next scan takes them — a **deferral, not a drop**: nothing is quarantined, errored, or left unaccounted for. See [*Per-tick poll ceilings*](#per-tick-poll-ceilings) for the number and when to raise it. `None`/`0` = unlimited. | | `validate_directory` | both | `false` | validate the directory **at startup** (#114): a missing/unusable dir reports the connection **`failed`** (ADR 0031) instead of the default deferral to run time. **No mkdir** — a merely-missing dir fails. **In:** a `leave` source validates read-only (a read-only share passes); `move`/`delete` also require write. **Out:** the target must already exist and accept a write, and is then **never created** — not at start, not on write (a delivery into a vanished dir fails retryably instead), and not by `POST /connections/{name}/test`. Left off (the default) the outbound target is still created on first write, but the creation is now logged as a `WARNING`. | | `processed_subdir` / `error_subdir` | in | `.processed` / `.error` | where read/failed files go | @@ -1055,7 +1055,7 @@ its own policy block below): charged on the handle the read opens, not on an earlier `stat()`: an oversize file is rejected by its handle's size **before** it is read into memory, and the read itself stops at the cap plus one byte, so a file that grows after it was listed is refused too (OOM / DoS guard, BACKLOG #2507). - `None`/`0` disables it. + `None`/`0` disables it, and the connection then logs a `WARNING` each time it is built (BACKLOG #2749). - **Links are refused at listing, read and move time.** The listing looks at each name without following a link there. A drop is read only if the opened handle is a regular file at the listed name inside the watch directory, reached through no symbolic link or junction: POSIX opens each path @@ -1312,7 +1312,7 @@ poll/write shape against a remote server, selected by an internal `protocol` set | `poll_seconds` | in | `5.0` | poll interval, a finite number of seconds above zero; anything else is refused at build, including at least zero, a negative value, NaN, infinity and text that is not a number (BACKLOG #2774). It is also the **settle window**: a file is read only once it lists at the same size, and the same modification time where the server lists one, as at the last poll that saw it (BACKLOG #2071, #2758), so every file waits at least one poll. The gate is always on and has no setting. On a server that lists no modification time (at least FTP without `MLSD`, an `MLSD` server without `modify`, and an SFTP server that omits the times) it reads the listed size alone, so there it cannot see a same-size rewrite, nor anything on a server that lists every file at size 0. | | `min_age_seconds` | in | `0.0` | **accepted but not honoured on a remote source today** — the connector never reads it. Only `File(...)` implements it; the settle gate on `poll_seconds` above, and the partner's own write-then-rename, are what guard against partial reads here. | | `after_read` | in | `move` | `move` (→ `processed_subdir`), `delete`, or `leave` (process **in place**, #142 — a durable dedup ledger keyed on a hash of the **full remote path** + size + the listed modification time where the server gives one ensures a left file is ingested once; see *Process-in-place* for what a server that lists no modification time misses) | -| `max_file_bytes` | in | `16 MiB` | **charged twice, and the second charge is the one that binds.** Before the retrieve, against the size the **server reported** in its own directory listing — an over-size entry is moved to `error_subdir` without being read. Then **during** the retrieve, against the **bytes actually read**: the download streams in 1 MiB chunks and is cut off at the first byte past the budget, so a share that lists a small file and then delivers an arbitrarily large body is refused mid-transfer rather than buffered whole (BACKLOG #1191). Either refusal quarantines the file to `error_subdir` and logs it — never a silent drop, and never left in place to be re-pulled every poll. `None`/`0` = unlimited, in both charges. | +| `max_file_bytes` | in | `16 MiB` | **charged twice, and the second charge is the one that binds.** Before the retrieve, against the size the **server reported** in its own directory listing — an over-size entry is moved to `error_subdir` without being read. Then **during** the retrieve, against the **bytes actually read**: the download streams in 1 MiB chunks and is cut off at the first byte past the budget, so a share that lists a small file and then delivers an arbitrarily large body is refused mid-transfer rather than buffered whole (BACKLOG #1191). Either refusal quarantines the file to `error_subdir` and logs it — never a silent drop, and never left in place to be re-pulled every poll. `None`/`0` = unlimited, in both charges, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `poll_max_files` | in | `500` | most files one poll will take. The rest stay on the share and the next poll takes them — a **deferral, not a drop**. Identical in shape and reasoning to the `File(...)` row; see [*Per-tick poll ceilings*](#per-tick-poll-ceilings). `None`/`0` = unlimited. | | `validate_directory` | both | `false` | validate `remote_dir` **at startup** (#114): unreachable/unusable reports the connection **`failed`** (ADR 0031) instead of deferring to run time. The probe is a **listing** — it never creates. **Out:** the upload dir is then never `ensure_dir`ed either, on send or by `POST /connections/{name}/test`. Instead every send first **lists** `remote_dir`, so the account needs list permission there even with `overwrite = true`, at one extra round trip per delivery. An upload into a vanished dir then fails **retryably** rather than dead-lettering on the partner's permanent no-such-dir; a credential refusal on that listing stops the lane. Left off (the default) the upload dir is still created on first send, but the creation is now logged as a `WARNING`. | | `processed_subdir` / `error_subdir` | in | `.processed` / `.error` | where read / failed files go | @@ -3581,9 +3581,9 @@ Facts that are easy to get wrong, stated plainly first: Noise in the same read as a completed frame starts nothing. A sender that holds a quiet socket open with them is closed at `max_frame_seconds`, even where `receive_timeout` is longer or off. - * On X12 the frame is the interchange, from its `ISA` to its `IEA`. X12 has no line-end allowance - yet, so a newline after the `IEA` that arrives in a read of its own starts the clock. That is a - known gap, not the intended rule. + * On X12 the frame is the interchange, from its `ISA` to the terminator of its `IEA` segment. Up + to two bytes directly after that terminator belong to the interchange when each is CR or LF, so + a newline after the `IEA` starts nothing, however late it arrives (vault BACKLOG #2938). * Time the engine withholds, for pacing or an intake pause, is not counted. Vault BACKLOG #2847 moved MLLP onto this rule. Its clock used to start only at a start byte. diff --git a/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md b/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md index e1f8759844..d211170fc1 100644 --- a/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md +++ b/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md @@ -94,6 +94,14 @@ processed_files( opportunistically at the end of a poll tick only when ≥1 new file was recorded, so a stable read-only share (nothing new) never churns the DB. +> **CORRECTION 2026-10-16 (BACKLOG #3151). The remote key above is out of date: since BACKLOG #2758 +> it also folds in the modification time the server lists, where it lists one.** The first bullet's +> reason, that a remote listing "carries no mtime", was wrong for at least SFTP and FTP `MLSD`. A +> listing with no mtime still keys on path + size. The `file_key` comment in the sketch above, +> `sha256(rel-path || mtime || size)`, describes the local key only. `RemoteFileSource._file_key` in +> `messagefoundry/transports/remotefile.py` is the source of record for the remote key, including +> which listings carry no mtime and what each key misses. + ### §3 Store-agnostic transport — a runner-injected ledger seam `transports/` must never import `store/`. The source therefore reaches the ledger through an injected diff --git a/messagefoundry/framing.py b/messagefoundry/framing.py index faaadcca3d..f1cb28abc7 100644 --- a/messagefoundry/framing.py +++ b/messagefoundry/framing.py @@ -91,6 +91,10 @@ class FrameEncodeError(ValueError): :class:`UnicodeEncodeError` quotes the character and carries the whole payload.""" +#: What :meth:`FrameCodec.neutralise` puts in place of a frame byte. +_SPACE = b" " + + def _body(payload: str | bytes, encoding: str) -> bytes: return payload.encode(encoding) if isinstance(payload, str) else bytes(payload) @@ -118,6 +122,12 @@ class FrameCodec: stateful streaming :class:`FrameDecoder`. The ``trailer`` (e.g. MLLP's CR) is **emitted** when framing but **not required** when decoding — a tolerant receiver treats any inter-frame bytes, including a stray trailer after the end delimiter, as noise to discard. + + ``start`` and ``end`` must differ, and neither may be ``0x20``, the space (BACKLOG #2900). + :meth:`neutralise` replaces each frame byte in a reply with a space, so a space frame byte would + be replaced by itself and the reply could not go as one frame; and a delivery holding a space + would be refused under ADR 0205 rule 1. A space trailer is allowed: the start byte is never a + space, so a space trailer cannot open a frame. """ start: int @@ -132,6 +142,12 @@ def __post_init__(self) -> None: raise ValueError(f"frame {name} must be a byte value in 0..255, got {value!r}") if self.start == self.end: raise ValueError("frame start and end delimiters must differ") + for name, value in (("start", self.start), ("end", self.end)): + if value == _SPACE[0]: # the class docstring says why + raise ValueError( + f"frame {name} must not be 0x20 (a space): a space is what replaces a frame " + "byte in a reply, and a space in a message would be taken as a frame byte" + ) def frame(self, payload: str | bytes, encoding: str = "utf-8") -> bytes: """Wrap a message: ``start payload end [trailer]``.""" @@ -167,7 +183,7 @@ def neutralise(self, body: bytes) -> bytes: comes back as is, uncopied; compare with ``!=`` to learn whether anything was replaced.""" if self.start not in body and self.end not in body: return body - return body.replace(bytes([self.start]), b" ").replace(bytes([self.end]), b" ") + return body.replace(bytes([self.start]), _SPACE).replace(bytes([self.end]), _SPACE) def frame_checked( self, payload: str | bytes, encoding: str = "utf-8", *, transport: str = "frame" @@ -184,8 +200,9 @@ def frame_checked( return self.frame(body) def frame_neutralised(self, payload: str | bytes, encoding: str = "utf-8") -> bytes: - """:meth:`frame` after :meth:`neutralise`: one frame for a reply, for any codec whose start - and end bytes are not the space. Raises :class:`FrameEncodeError` as :meth:`frame_checked`.""" + """:meth:`frame` after :meth:`neutralise`, so the framed body holds neither frame byte: no + codec may have the space as its start or end byte (see the class docstring). Raises + :class:`FrameEncodeError` as :meth:`frame_checked`.""" return self.frame(self.neutralise(_encoded(payload, encoding, "reply"))) def decoder(self, max_frame_bytes: int | None = None) -> FrameDecoder: diff --git a/messagefoundry/parsing/x12/interchange.py b/messagefoundry/parsing/x12/interchange.py index 8c4283a6b4..95b1d0c518 100644 --- a/messagefoundry/parsing/x12/interchange.py +++ b/messagefoundry/parsing/x12/interchange.py @@ -43,6 +43,17 @@ # A segment that is the IEA trailer, after cosmetic whitespace. Matched in place with pos/endpos, so the # check never copies a segment, however long. _IEA_SEGMENT = re.compile(rb"[ \t\r\n\x0b\x0c]*IEA") +#: Line-end bytes a sender may put after an interchange's closing terminator, such as a newline after +#: ``IEA*...~``. Up to :data:`_FRAME_TAIL_MAX` of them directly after a completed interchange are its +#: ending, however reads split them (vault BACKLOG #2938). The cap keeps a peer from trickling them +#: to hold a listener's frame deadline off: past it they are inter-interchange noise like any byte. +#: The same rule as ``messagefoundry.framing.FrameDecoder``'s, written as a local copy because +#: ``parsing`` may not import ``messagefoundry.framing`` (its import allowlist in +#: tests/test_dependency_boundaries.py). So the two can drift; the x12 and MLLP rows of +#: ``test_line_end_bytes_ending_a_frame_do_not_start_the_frame_deadline`` in +#: tests/test_listener_admission.py pin that both allow two line ends and cut on a third. +_FRAME_TAIL_BYTES = b"\r\n" +_FRAME_TAIL_MAX = 2 def split(raw: str) -> list[str]: @@ -101,6 +112,15 @@ def __init__(self, max_interchange_bytes: int | None = None) -> None: # received begins; ``_scan`` is where the terminator search picks up, at or after it. self._seg_start = 0 self._scan = 0 + #: Line-end bytes the last completed interchange may still take as its ending; 0 once any + #: other byte has followed it. See :data:`_FRAME_TAIL_BYTES`. + self._tail_left = 0 + #: Whether the most recent :meth:`feed` held nothing but line-end bytes ending the + #: interchange before it. Such a read finishes that interchange rather than starting anything, + #: which is how a listener's frame deadline tells it from inter-interchange noise (vault + #: BACKLOG #2938). Set when :meth:`feed` is called, before its generator is iterated. It only + #: observes: the bytes still go to the buffer, so the noise handling is unchanged. + self.trailer_only = False @property def in_frame(self) -> bool: @@ -110,6 +130,24 @@ def in_frame(self) -> bool: return self._buf.startswith(b"ISA") def feed(self, data: bytes) -> Iterator[bytes]: + """Yield each interchange completed by ``data``, carrying any partial one to the next call. + + :attr:`trailer_only` is decided here, eagerly, so it never describes an earlier read; the + reassembly itself still runs lazily as the generator is iterated. The allowance it reads is + set as an interchange is taken, so drain one call's generator before the next call, as the + listener does; a caller that stops early reads the next call's flag against a stale + allowance. + """ + self.trailer_only = bool(data) and _only_line_ends(data, self._tail_left) + # Any other byte ends the allowance: what follows it is not the interchange's ending. An + # empty read holds no byte, so it leaves the allowance as it was, as FrameDecoder does. + if data and not self.trailer_only: + self._tail_left = 0 + else: + self._tail_left -= len(data) + return self._frames(data) + + def _frames(self, data: bytes) -> Iterator[bytes]: self._buf.extend(data) while True: frame = self._take_one() @@ -163,6 +201,10 @@ def _take_one(self) -> bytes | None: frame = bytes(buf[:end]) del buf[:end] self._seg_start = self._scan = 0 + # What the read held past the terminator: line ends alone are owed to this + # interchange, so the allowance left is what they have not used yet. + tail = _only_line_ends(buf, _FRAME_TAIL_MAX) + self._tail_left = _FRAME_TAIL_MAX - len(buf) if tail else 0 return frame self._seg_start = self._scan = idx + len(terminator) @@ -191,6 +233,12 @@ def _check_cap(self) -> None: ) +def _only_line_ends(data: bytes | bytearray, limit: int) -> bool: + """Whether ``data`` is at most ``limit`` bytes, every one a line end. The length is checked + first, so a large buffer is never copied by the strip.""" + return len(data) <= limit and not data.strip(_FRAME_TAIL_BYTES) + + def check_integrity(raw: str) -> list[str]: """Structural integrity tie-out for one interchange (a pure helper, opt-in). diff --git a/messagefoundry/store/store.py b/messagefoundry/store/store.py index 3468e5035a..3427cb9f62 100644 --- a/messagefoundry/store/store.py +++ b/messagefoundry/store/store.py @@ -6660,8 +6660,9 @@ def _dead_target_pairs(rows: Iterable[tuple[Any, Any]]) -> list[tuple[str, str]] -- files another system owns can be polled without moving/deleting them and a left file is ingested -- ONCE. Shaped like `delivered_keys`/`resend_log`: HASHES + IDS ONLY, no body/PHI, so it is stored in -- the clear (NOT part of the `_cipher` seam). `file_key` is a SHA-256 of the file's identity --- (relative-path+mtime+size local / full-remote-path+size remote — the PATH is folded in, not just the --- basename, so two same-named files under different subdirs stay distinct and are BOTH ingested) — a +-- (relative-path+mtime+size local / full-remote-path+size+the listed mtime remote, the mtime only where +-- the server lists one, BACKLOG #2758 — the PATH is folded in, not just the basename, so two +-- same-named files under different subdirs stay distinct and are BOTH ingested) — a -- DERIVED id, NEVER a cleartext path (a filename/path can embed an MRN) and never logged at INFO+. -- `channel_id` is the inbound connection name (config metadata, non-PHI) — scopes the dedup + the -- age/count prune per connection. The row is INSERTed AFTER the file's message(s) emit successfully (the diff --git a/messagefoundry/transports/admission.py b/messagefoundry/transports/admission.py index c229595c8d..7bc75aed07 100644 --- a/messagefoundry/transports/admission.py +++ b/messagefoundry/transports/admission.py @@ -119,9 +119,9 @@ class Refusal: class ListenerAdmission: """The pre-read admission decision for one socket listener, and the counts it rests on. - Call :meth:`check` first. ``None`` means admit: call :meth:`admit`, and pair it with exactly one - :meth:`release` in a ``finally``. A :class:`Refusal` means close the socket unread; nothing was - taken, so nothing is released. + Call :meth:`check` first. ``None`` means admit: call :meth:`admit`, keep the epoch it returns, + and pair it with exactly one :meth:`release` of that epoch in a ``finally``. A :class:`Refusal` + means close the socket unread; nothing was taken, so nothing is released. ``max_connections`` counts sockets on this listener. ``max_connections_per_host`` counts sockets from one peer address, so one address cannot take the whole listener. It keys on the source IP, @@ -152,6 +152,10 @@ def __init__( #: Hosts already warned about their per-host cap this episode, cleared with the count above. self.host_capacity_warned: set[str] = set() self.refusal_log = RefusalLog() + #: Which run of the listener the per-host tables describe. :meth:`admit` hands it to the + #: connection and :meth:`reset` advances it, so a release from an earlier run can be told + #: from one the tables still count (vault BACKLOG #2939). + self._epoch = 0 def check(self, writer: asyncio.StreamWriter, peer_host: str | None) -> Refusal | None: """Decide whether to admit this connection. Logs a refusal, throttled; emits no event.""" @@ -173,17 +177,29 @@ def check(self, writer: asyncio.StreamWriter, peer_host: str | None) -> Refusal return _AT_HOST_CAPACITY return None - def admit(self, peer_host: str | None) -> None: - """Take one connection slot, globally and (when the peer has an address) for that host.""" + def admit(self, peer_host: str | None) -> int: + """Take one connection slot, globally and (when the peer has an address) for that host. + + Returns the epoch to hand back to :meth:`release`.""" self.active += 1 if peer_host is not None: self.per_host[peer_host] = self.per_host.get(peer_host, 0) + 1 + return self._epoch - def release(self, peer_host: str | None) -> None: + def release(self, peer_host: str | None, epoch: int) -> None: """Give both slots back. One place for both counters, so they cannot drift: a per-host count - left behind by a missed release would lock that peer out until restart.""" + left behind by a missed release would lock that peer out until restart. + + ``epoch`` is what :meth:`admit` returned, and it is required, so a call site cannot leave it + out. A release from an earlier epoch is a connection that outlived a stop: :meth:`reset` + already dropped its per-host slot, and the table now counts the restarted listener's + connections, so it touches neither that table nor the warned set (vault BACKLOG #2939). + It still gives back ``active``, which :meth:`reset` leaves alone: that socket was open, and + counted, until now, whichever run admitted it. The per-host cap is not exact across a + restart: while such a straggler is still open, its host may hold one socket over + ``max_connections_per_host`` for it, since the reset already dropped its slot.""" self.active -= 1 - if peer_host is None: + if peer_host is None or epoch != self._epoch: return remaining = self.per_host.get(peer_host, 0) - 1 if remaining > 0: @@ -199,9 +215,13 @@ def reset(self) -> None: A no-op once the listener has started again: a stop() that finishes after a restart must not wipe the counts of connections the restarted listener has admitted. Call it from a - ``finally`` so a cancelled stop() still clears the tables.""" + ``finally`` so a cancelled stop() still clears the tables. + + Clearing advances the epoch, so a connection admitted before it that releases later gives + back no per-host slot it no longer holds.""" if not self.stopping: return + self._epoch += 1 self.per_host.clear() self.host_capacity_warned.clear() @@ -357,11 +377,13 @@ def after_read(self, *, in_frame: bool, decoded: int, trailer_only: bool) -> Non ``in_frame`` would never reach a peer trickling them. So any bytes must lead to a completed frame within ``max_frame_seconds`` of read time; the engine's own waits are withheld. - ``trailer_only`` (:attr:`~messagefoundry.framing.FrameDecoder.trailer_only`) is the one - exception: a read holding only the line-end bytes that end the frame before it, such as - MLLP's CR arriving after its FS, belongs to that completed frame and starts nothing (vault - BACKLOG #2847). Without it a healthy peer that went quiet after such a read would be cut - here. It is required, so a new caller cannot leave it out by accident. + ``trailer_only`` (:attr:`~messagefoundry.framing.FrameDecoder.trailer_only`, or + :attr:`~messagefoundry.parsing.x12.interchange.X12FrameReader.trailer_only` for X12) is the + one exception: a read holding only the line-end bytes that end the frame before it, such as + MLLP's CR arriving after its FS or a newline after an X12 ``IEA``, belongs to that completed + frame and starts nothing (vault BACKLOG #2847, #2938). Without it a healthy peer that went + quiet after such a read would be cut here. It is required, so a new caller cannot leave it + out by accident. """ if decoded: self.opened_at = time.monotonic() if in_frame else None diff --git a/messagefoundry/transports/file.py b/messagefoundry/transports/file.py index 9a34122e98..a5e4fdde14 100644 --- a/messagefoundry/transports/file.py +++ b/messagefoundry/transports/file.py @@ -677,6 +677,13 @@ def __init__(self, config: Source) -> None: self.min_age_seconds: float = float_setting( s.get("min_age_seconds", 0.0), knob="min_age_seconds", transport="file source" ) + # Zero is the shipped default (no extra wait), so this is not positive_seconds; only a negative + # is refused, as a typo. The scan's `> 0` test would read it as zero (BACKLOG #3151). + if self.min_age_seconds < 0: + raise ValueError( + "file source min_age_seconds must be a number of seconds, zero or more " + "(value withheld)" + ) self.after_read: str = s.get("after_read", "move") # "move" | "delete" | "leave" (#142) if self.after_read not in ("move", "delete", "leave"): raise ValueError( @@ -762,6 +769,15 @@ def __init__(self, config: Source) -> None: # service-account identity, byte-identical. On a non-Windows host a configured credential # raises CredentialUnsupportedError here (a build error), never a silent no-op. self._cred_ctx = _build_credential_context(s) + if self.max_file_bytes is None: + # Off is allowed but never silent (BACKLOG #2749). Logged after every check in this + # constructor, so a source it refuses has not logged it. The line names the setting, + # never its value. + logger.warning( + "file source %s: max_file_bytes is off, so a drop is read whole with no byte " + "ceiling; set a positive value to bound it", + config.name or self.directory, + ) async def _run_fs(self, fn: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: """Run a blocking filesystem callable off the event loop — under the alternate credential (on a diff --git a/messagefoundry/transports/http_listener.py b/messagefoundry/transports/http_listener.py index d767c580e8..7c34b71aac 100644 --- a/messagefoundry/transports/http_listener.py +++ b/messagefoundry/transports/http_listener.py @@ -775,6 +775,15 @@ def __init__(self, config: Source) -> None: # connections and bound the wait (mirrors MLLPSource, review H-2 / #55). self._clients: set[asyncio.StreamWriter] = set() self._client_tasks: set[asyncio.Task[None]] = set() + if self.max_body_bytes is None: + # Turning the body cap off is documented, so it is allowed, but never silent (BACKLOG + # #2749, as FileSource does for its caps). Logged after every check in this constructor, + # so a source it refuses has not logged it. The line names the setting, never its value. + logger.warning( + "HTTP source %s: max_body_bytes is off, so a request body of any declared size is " + "read whole; set a positive value to bound it", + config.name or f"{self.host}:{self.port}", + ) async def start( self, handler: InboundHandler, *, leader_gate: Callable[[], bool] | None = None @@ -1078,7 +1087,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW ), ) return # pre-ingress refusal — nothing accepted, so nothing to count - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) established = True # set with the admit, so the finally below always releases it await self._emit_event("established", peer_host=peer_host) failed = await self._serve_one(reader, writer, peer_host=peer_host) @@ -1094,7 +1103,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW await self._write_safely(writer, build_response(500, '{"error":"internal error"}')) finally: if established: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) self._clients.discard(writer) if task is not None: self._client_tasks.discard(task) diff --git a/messagefoundry/transports/mllp.py b/messagefoundry/transports/mllp.py index 24ab2138bf..f1f4471b3a 100644 --- a/messagefoundry/transports/mllp.py +++ b/messagefoundry/transports/mllp.py @@ -2192,12 +2192,12 @@ async def stop(self) -> None: # at the next promotion) may not run its `finally` before the restart, so a count it # holds would survive into the restarted listener, and `release`'s own docstring names # what that costs: a per-host count left behind by a missed release locks that peer - # out. The limit, stated so it is not over-read: a straggler that releases AFTER the - # restart has re-admitted its host decrements the new count, so the per-host cap can - # undercount by one per such straggler while that host's newer connections stay open. - # Tying a release to the run - # that admitted it would close that, and is a separate change. `active` keeps its - # existing behaviour: the global cap that predates this row reads it. + # out. A straggler that releases AFTER the restart has re-admitted its host carries the + # epoch of the run that admitted it, and `reset` advances the epoch, so that release + # leaves the restarted listener's per-host count alone rather than undercounting it + # (vault BACKLOG #2939). `active` still drops on it: that socket was open until then, + # and the global cap that predates this row reads it. What the epoch leaves open while + # a straggler is still live is stated once, in `ListenerAdmission.release`. self._admission.reset() # Now that no client handlers are in flight, this should complete promptly — but on the Windows # ProactorEventLoop a still-pending overlapped accept/read can make wait_closed() never return, @@ -2374,7 +2374,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # refused (closed in the outer finally) - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) # The `established` emit sits INSIDE the try whose finally releases the slot. It is # fail-soft against `Exception` but not against `CancelledError`, and stop() cancels # straggling client tasks — so a cancellation delivered here used to escape past @@ -2500,7 +2500,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/messagefoundry/transports/remotefile.py b/messagefoundry/transports/remotefile.py index 30e2a7bdfc..628e1f8c98 100644 --- a/messagefoundry/transports/remotefile.py +++ b/messagefoundry/transports/remotefile.py @@ -47,7 +47,8 @@ the body *before* an ingress row exists, so no admission bound further down the pipeline can see it. The refusal is a content refusal, not a transient one — the file is quarantined to ``error_subdir`` and logged, exactly as an over-listed-size or content-sniff reject is. ``max_file_bytes=0`` disables -both charges (unbounded), which is the operator's explicit choice. +both charges (unbounded), which is the operator's explicit choice, and the source logs a WARNING +saying so each time it is built (BACKLOG #2749). **Idempotency.** Delivery is at-least-once (an upload may re-send) and a poll may re-emit a file that was handled but not yet marked, so downstream consumers **must** tolerate duplicates. @@ -2453,6 +2454,15 @@ def __init__(self, config: Source) -> None: self._other_form_warned_at: float | None = None self._stop = asyncio.Event() self._task: asyncio.Task[None] | None = None + if self._max_file_bytes is None: + # Turning the cap off is documented, so it is allowed, but never silent (BACKLOG #2749, + # as FileSource does for its caps). Logged after every check in this constructor, so a + # source it refuses has not logged it. The line names the setting, never its value. + logger.warning( + "REMOTEFILE source %s: max_file_bytes is off, so a retrieve is read whole with no " + "byte ceiling; set a positive value to bound it", + self._connection or _where(self._host, self._remote_dir), + ) async def start( self, handler: InboundHandler, *, leader_gate: Callable[[], bool] | None = None diff --git a/messagefoundry/transports/tcp.py b/messagefoundry/transports/tcp.py index f77d439bb7..f596f30600 100644 --- a/messagefoundry/transports/tcp.py +++ b/messagefoundry/transports/tcp.py @@ -638,7 +638,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # closed in the outer finally - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) try: # Inside the try whose finally releases the slot, so a cancellation delivered at # this await cannot leak a per-host count. @@ -718,7 +718,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/messagefoundry/transports/x12.py b/messagefoundry/transports/x12.py index b3140747d2..61d8239e73 100644 --- a/messagefoundry/transports/x12.py +++ b/messagefoundry/transports/x12.py @@ -777,7 +777,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # closed in the outer finally - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) try: # Inside the try whose finally releases the slot, so a cancellation delivered at # this await cannot leak a per-host count. @@ -851,15 +851,18 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW ) break # Reached on the success path alone: every arm above breaks or raises. - # X12FrameReader reports no end-of-frame bytes yet, so a newline after the IEA - # segment terminator that arrives in a read of its own starts the clock. That - # gap is open, not intended (vault BACKLOG #2847 review). - clock.after_read(in_frame=decoder.in_frame, decoded=decoded, trailer_only=False) + # A newline after the IEA segment terminator that arrives in a read of its own + # ends that interchange and starts no clock (vault BACKLOG #2938). + clock.after_read( + in_frame=decoder.in_frame, + decoded=decoded, + trailer_only=decoder.trailer_only, + ) except OSError as exc: failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/tests/test_inbound_http_source.py b/tests/test_inbound_http_source.py index 7eba06389c..1e99f0c77e 100644 --- a/tests/test_inbound_http_source.py +++ b/tests/test_inbound_http_source.py @@ -505,6 +505,57 @@ async def test_read_request_rejects_declared_oversize() -> None: assert exc.value.status == 413 and exc.value.kind == "frame_oversize" +def _http_source(**over: Any) -> HttpSource: + ic = build_inbound_connection("IB_HTTP", Http(port=0, **over), router="r") + return HttpSource(_source_config(ic, "127.0.0.1", {})) + + +def _body_cap_off_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + r + for r in caplog.records + if r.levelno == logging.WARNING and "max_body_bytes is off" in r.getMessage() + ] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_http_source_warns_once_when_max_body_bytes_is_off( + off: object, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #2749: off is documented and allowed, but never silent. The line names the + connection and the setting, and withholds the value. + + Red mutation: drop the warning from ``HttpSource.__init__``. No record is found.""" + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.http_listener"): + src = _http_source(max_body_bytes=off) + assert src.max_body_bytes is None + [record] = _body_cap_off_warnings(caplog) + assert "IB_HTTP" in record.getMessage() + + +async def test_http_source_default_max_body_bytes_warns_nothing_and_still_refuses_oversize( + store: MessageStore, caplog: pytest.LogCaptureFixture +) -> None: + """The control for the warning above, through the running listener: the shipped default says + nothing, and a body declared one byte past it is refused ``413`` before a body byte is sent.""" + ic = build_inbound_connection( + "IB_HTTP", Http(port=0), router="r", content_type=ContentType.TEXT + ) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.http_listener"): + src = await _start_source(store, ic) + try: + assert _body_cap_off_warnings(caplog) == [] + declared = DEFAULT_MAX_BODY_BYTES + 1 + head = f"POST /ingest HTTP/1.1\r\nHost: localhost\r\nContent-Length: {declared}\r\n\r\n" + resp = await _http(src.sockport, raw_override=head.encode("ascii")) + finally: + await src.stop() + assert resp.status == 413 + cur = await store._db.execute("SELECT COUNT(*) AS n FROM messages") + row = await cur.fetchone() + assert row is not None and row["n"] == 0 # refused before any ingress row + + async def test_read_request_rejects_chunked() -> None: raw = b"POST /x HTTP/1.1\r\nHost: h\r\nTransfer-Encoding: chunked\r\n\r\n" with pytest.raises(HttpRequestError) as exc: diff --git a/tests/test_listener_admission.py b/tests/test_listener_admission.py index c43ae346f4..93c477ddaa 100644 --- a/tests/test_listener_admission.py +++ b/tests/test_listener_admission.py @@ -433,6 +433,11 @@ async def body(port: int) -> None: ("tcp", b"\r", False), ("mllp", b"Z", True), # CONTROL: any other byte after a completed frame starts the clock ("mllp", b"\r\n\r", True), # CONTROL: line ends past the two-byte allowance are noise + # vault BACKLOG #2938: the newline after an X12 IEA terminator takes the same allowance. + ("x12", b"\r\n", False), + ("x12", b"\n", False), # the CR arrived with the IEA, so the LF is the second byte + ("x12", b"Z", True), # CONTROL: a byte that is not a line end starts the clock + ("x12", b"\r\n\r", True), # CONTROL: three line ends are past the allowance ], ) async def test_line_end_bytes_ending_a_frame_do_not_start_the_frame_deadline( @@ -452,7 +457,8 @@ async def handler(raw: bytes) -> str | None: framing = {"framing": "mllp"} if kind == "tcp" else {} source = _build(kind, max_frame_seconds=0.3, receive_timeout=5.0, **framing) source.on_connection_event = events - whole = _whole_frame("mllp") + # X12's frame is the interchange through its IEA terminator; the CR here is its first line end. + whole = _whole_frame("x12") + b"\r" if kind == "x12" else _whole_frame("mllp") # The `\n` row sends the CR with the frame, so the LF is the second line-end byte. first = whole if after == b"\n" else whole[:-1] @@ -597,21 +603,23 @@ def test_the_refusal_log_defaults_match_the_dicom_server() -> None: # --- Stopping, and the slot accounting ------------------------------------------------------------- +class _PeerWriter: + """The one writer method admission reads, for a peer at ``192.0.2.7``.""" + + def get_extra_info(self, name: str, default: object = None) -> object: + return ("192.0.2.7", 4000) if name == "peername" else default + + def test_admission_refuses_while_stopping_and_releases_what_it_admits() -> None: gate = ListenerAdmission( transport="TCP", max_connections=2, max_connections_per_host=1, source_ip_allowlist=None ) - - class _W: - def get_extra_info(self, name: str, default: object = None) -> object: - return ("192.0.2.7", 4000) if name == "peername" else default - - w: Any = _W() + w: Any = _PeerWriter() assert gate.check(w, "192.0.2.7") is None - gate.admit("192.0.2.7") + epoch = gate.admit("192.0.2.7") refusal = gate.check(w, "192.0.2.7") assert refusal is not None and refusal.reason == "max_connections_per_host" - gate.release("192.0.2.7") + gate.release("192.0.2.7", epoch) assert gate.active == 0 and gate.per_host == {} assert gate.host_capacity_warned == set() gate.stopping = True @@ -619,6 +627,58 @@ def get_extra_info(self, name: str, default: object = None) -> object: assert refusal is not None and refusal.kind is None, "a stop refusal emits no event" +def test_a_release_from_before_a_restart_leaves_the_restarted_count_alone() -> None: + """vault BACKLOG #2939: a connection that outlives its listener's stop() and releases after the + restart has re-admitted its host must not take a slot from the new count. Without the epoch, the + straggler's release would drop the host to one below the cap and let a third connection in.""" + host = "192.0.2.7" + w: Any = _PeerWriter() + gate = ListenerAdmission( + transport="TCP", max_connections=10, max_connections_per_host=2, source_ip_allowlist=None + ) + straggler = gate.admit(host) + gate.stopping = True # stop() begins; the straggler's task outlives it + gate.reset() + gate.stopping = False # the same instance restarts + assert gate.check(w, host) is None + current = gate.admit(host) + assert gate.check(w, host) is None + gate.admit(host) + refused = gate.check(w, host) + assert refused is not None and refused.reason == "max_connections_per_host" + assert host in gate.host_capacity_warned + gate.release(host, straggler) + assert gate.per_host == {host: 2}, "a pre-restart release took a slot from the restarted run" + refused = gate.check(w, host) + assert refused is not None and refused.reason == "max_connections_per_host" + assert host in gate.host_capacity_warned, "the warned set belongs to the restarted run too" + # `active` counts sockets, and the straggler's was open until now, whichever run admitted it. + assert gate.active == 2 + # CONTROL: a release from the current run still gives its slot back. + gate.release(host, current) + assert gate.per_host == {host: 1} + assert gate.check(w, host) is None + + +def test_a_release_still_counts_when_a_late_stop_skipped_the_reset() -> None: + """The epoch advances only when reset() really clears the tables. A stop() that finishes after + the restart skips its reset, so the tables still hold the earlier connection's slot, and its + release must give that slot back, or the host would stay one short of its cap for good.""" + host = "192.0.2.7" + gate = ListenerAdmission( + transport="TCP", max_connections=None, max_connections_per_host=2, source_ip_allowlist=None + ) + earlier = gate.admit(host) + gate.stopping = True + gate.stopping = False # restarted before the stop() reached its reset + gate.reset() # the late stop(): a no-op now + gate.admit(host) + assert gate.per_host == {host: 2} + gate.release(host, earlier) + assert gate.per_host == {host: 1} + assert gate.active == 1 + + @pytest.mark.parametrize("kind", ["tcp", "x12", "http"]) async def test_a_connection_reaching_a_stopping_listener_is_refused_unread(kind: str) -> None: events = _Events() @@ -705,6 +765,54 @@ async def failing_wait(fs: Any, *args: Any, **kwargs: Any) -> Any: await asyncio.wait_for(source.stop(), timeout=10.0) +@pytest.mark.parametrize("kind", ["mllp", "tcp", "x12"]) +async def test_a_straggler_releasing_after_a_restart_leaves_the_new_per_host_count( + kind: str, +) -> None: + """vault BACKLOG #2939, end to end through the listener's own admit and release. A client held + in its handler outlives a stop() cut short; the same instance restarts and admits the host + again; then the straggler releases. The restarted run's count must still hold its connection. + This pins the MLLP, raw-TCP and X12 listeners passing on the epoch their admit returned. The + HTTP listener's handler has another shape and is not driven here.""" + entered = asyncio.Event() + release = asyncio.Event() + + async def handler(raw: bytes) -> str: + entered.set() + await release.wait() + return build_ack(raw, code="AA") + + source = _build(kind) + await source.start(handler) + gate = source._admission # type: ignore[attr-defined] + writers: list[asyncio.StreamWriter] = [] + try: + _reader, old = await asyncio.open_connection("127.0.0.1", source.sockport) # type: ignore[attr-defined] + writers.append(old) + old.write(_whole_frame(kind, HL7)) + await old.drain() + await asyncio.wait_for(entered.wait(), 3.0) + stopping = asyncio.ensure_future(source.stop()) + await asyncio.sleep(0.1) # inside the client-task grace wait + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + assert gate.per_host == {} # the reset ran; the straggler is still in its handler + await source.start(handler) # the same instance restarts + _reader, new = await asyncio.open_connection("127.0.0.1", source.sockport) # type: ignore[attr-defined] + writers.append(new) + assert await _until(lambda: gate.per_host == {"127.0.0.1": 1}) + assert gate.active == 2 + release.set() + assert await _until(lambda: gate.active == 1), "the straggler never released" + assert gate.per_host == {"127.0.0.1": 1}, "the straggler took the restarted run's slot" + finally: + release.set() + for writer in writers: + await _close(writer) + await asyncio.wait_for(source.stop(), timeout=10.0) + + # --- A normal frame still flows, and is answered, with every new bound at its default ------------- diff --git a/tests/test_remotefile_transport.py b/tests/test_remotefile_transport.py index 6f87965778..df3e30ab39 100644 --- a/tests/test_remotefile_transport.py +++ b/tests/test_remotefile_transport.py @@ -853,6 +853,48 @@ async def test_source_zero_max_file_bytes_keeps_the_retrieve_unbounded( assert len(h.bodies) == 1 # delivered whole — the operator disabled the cap +def _file_cap_off_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + r + for r in caplog.records + if r.levelno == logging.WARNING and "max_file_bytes is off" in r.getMessage() + ] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_source_warns_once_when_max_file_bytes_is_off( + off: object, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #2749: off is documented and allowed, but never silent. The line names where the + source polls and the setting, and withholds the value. + + Red mutation: drop the warning from ``RemoteFileSource.__init__``. No record is found.""" + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + src = _src(monkeypatch, _FakeClient(), max_file_bytes=off) + assert src._max_file_bytes is None + [record] = _file_cap_off_warnings(caplog) + assert "sftp.example.com:/in" in record.getMessage() + + +async def test_source_default_max_file_bytes_warns_nothing_and_still_refuses_oversize( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """The control for the warning above: the shipped default says nothing, and it still refuses a + file listed one byte past it, without retrieving it.""" + name = "/in/big.hl7" + client = _FakeClient(files={name: b"x"}, sizes={name: DEFAULT_MAX_FILE_BYTES + 1}) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + src = _src(monkeypatch, client) + assert _file_cap_off_warnings(caplog) == [] + h = _RecordingHandler() + src._handler = h + await _settle(src) + await src._poll_once() + assert h.bodies == [] + assert not any(op == "retrieve" for op, _ in client.ops) + assert "/in/.error/big.hl7" in client.files + + def test_sftp_retrieve_stops_reading_past_the_budget(monkeypatch: pytest.MonkeyPatch) -> None: """The SFTP backend reads in chunks and refuses mid-transfer, so a hostile body is never buffered whole. ``read_total`` proves the bytes were never pulled.""" diff --git a/tests/test_tcp_transport.py b/tests/test_tcp_transport.py index 1acb59987a..436eeab1c0 100644 --- a/tests/test_tcp_transport.py +++ b/tests/test_tcp_transport.py @@ -215,6 +215,47 @@ def test_codec_validates_bytes_and_distinct_delimiters() -> None: codec_for(None) # neither preset nor bytes +@pytest.mark.parametrize("role", ["start", "end"]) +def test_codec_refuses_a_space_start_or_end_byte(role: str) -> None: + # BACKLOG #2900. The refusal names the setting, directly, through the settings and at build. + def frame_bytes(byte: int) -> dict[str, int]: + return {"start": 0x02, "end": 0x03, role: byte} + + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + FrameCodec(**frame_bytes(0x20)) + with pytest.raises( + ValueError, match=f"TCP framing misconfigured: frame {role} must not be 0x20" + ): + tcp_mod._codec_from_settings(dict(frame_bytes(0x20))) + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + _source(framing=None, **frame_bytes(0x20)) + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + _dest(9, framing=None, **frame_bytes(0x20)) + # Control: the byte after the space still builds, directly, through the settings and at build. + built = FrameCodec(**frame_bytes(0x21)) + assert getattr(built, role) == 0x21 + assert tcp_mod._codec_from_settings(dict(frame_bytes(0x21))) == built + assert _source(framing=None, **frame_bytes(0x21)).codec == built + assert _dest(9, framing=None, **frame_bytes(0x21)).codec == built + + +def test_codec_allows_a_space_trailer() -> None: + # The trailer is not a frame byte to a decoder, so a space trailer is not refused. + codec = FrameCodec(start=0x02, end=0x03, trailer=0x20) + assert codec.frame_neutralised("A\x02B\x03C") == b"\x02A B C\x03 " + assert tcp_mod._codec_from_settings({"start": 2, "end": 3, "trailer": 0x20}) == codec + + +@pytest.mark.parametrize("trailer", [None, 0x20]) +def test_a_neutralised_reply_decodes_as_one_frame(trailer: int | None) -> None: + # The property behind the 0x20 refusal, for a codec that may be built: a reply holding both + # frame bytes still decodes as exactly one payload with no frame left open. + codec = FrameCodec(start=0x21, end=0x03, trailer=trailer) + decoder = codec.decoder() + assert list(decoder.feed(codec.frame_neutralised("A!B\x03C"))) == [b"A B C"] + assert not decoder.in_frame + + # --- MLLP unchanged after refactor onto the shared codec --------------------- diff --git a/tests/test_transports.py b/tests/test_transports.py index 41ecba95ae..6a13cb3d24 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -1749,11 +1749,11 @@ def test_file_source_member_cap_refuses_zero_and_bad_values(bad: object) -> None FileSource(Source(type=ConnectorType.FILE, settings=settings)) -def _decompressed_cap_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: +def _cap_off_warnings(caplog: pytest.LogCaptureFixture, knob: str) -> list[logging.LogRecord]: return [ r for r in caplog.records - if r.levelno == logging.WARNING and "max_decompressed_bytes is off" in r.getMessage() + if r.levelno == logging.WARNING and f"{knob} is off" in r.getMessage() ] @@ -1766,7 +1766,7 @@ def test_file_source_warns_once_when_the_decompressed_byte_cap_is_off( with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): src = FileSource(Source(type=ConnectorType.FILE, name="IB_DEMO_ADT", settings=settings)) assert src.max_decompressed_bytes is None - [record] = _decompressed_cap_warnings(caplog) + [record] = _cap_off_warnings(caplog, "max_decompressed_bytes") assert "IB_DEMO_ADT" in record.getMessage() @@ -1784,7 +1784,66 @@ def test_file_source_does_not_warn_while_the_decompressed_byte_cap_holds( # Without decompress the cap is never read, so turning it off changes nothing to warn about. with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): FileSource(Source(type=ConnectorType.FILE, settings={"directory": "x", **settings})) - assert _decompressed_cap_warnings(caplog) == [] + assert _cap_off_warnings(caplog, "max_decompressed_bytes") == [] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_file_source_warns_once_when_max_file_bytes_is_off( + off: object, caplog: pytest.LogCaptureFixture +) -> None: # #2749 + # Off is documented and allowed, but never silent. The line names the connection and the setting. + # Red mutation: drop the warning from FileSource.__init__. No record is found. + settings = {"directory": "x", "max_file_bytes": off} + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + src = FileSource(Source(type=ConnectorType.FILE, name="IB_DEMO_ADT", settings=settings)) + assert src.max_file_bytes is None + [record] = _cap_off_warnings(caplog, "max_file_bytes") + assert "IB_DEMO_ADT" in record.getMessage() + + +async def test_file_source_default_max_file_bytes_warns_nothing_and_still_refuses_oversize( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: # #2749 + # The control for the warning above: the shipped default says nothing, and a drop one byte past it + # is still quarantined unread. + inbox = tmp_path / "in" + inbox.mkdir() + (inbox / ".error").mkdir() # start() would make it; _scan_once is driven directly here + with (inbox / "big.hl7").open("wb") as fh: + fh.truncate(DEFAULT_MAX_FILE_BYTES + 1) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + src = _file_source_for(inbox) + assert _cap_off_warnings(caplog, "max_file_bytes") == [] + received: list[bytes] = [] + + async def handler(raw: bytes) -> None: + received.append(raw) + + src._handler = handler + await src._scan_once() # the settle poll (BACKLOG #1811): records the stat, reads nothing + await src._scan_once() + assert received == [] + assert (inbox / ".error" / "big.hl7").exists() + + +@pytest.mark.parametrize("bad", [-1, -0.5, "-1", "-0.5"]) +def test_file_source_refuses_a_negative_min_age_seconds(bad: object) -> None: # #3151 + # Red mutation: drop the negative check from FileSource.__init__. Every case builds. + settings = {"directory": "x", "min_age_seconds": bad} + refusal = ( + "file source min_age_seconds must be a number of seconds, zero or more (value withheld)" + ) + with pytest.raises(ValueError, match=f"^{re.escape(refusal)}$"): + FileSource(Source(type=ConnectorType.FILE, settings=settings)) + + +@pytest.mark.parametrize("ok", ["unset", 0, 0.0, "0", "-0", 2.5, "3600"]) +def test_file_source_accepts_zero_or_a_positive_min_age_seconds(ok: object) -> None: # #3151 + # The control for the refusal above: unset (the shipped default of zero), a zero in any exact + # spelling, a negative zero, and a positive value all build. + settings = {"directory": "x"} if ok == "unset" else {"directory": "x", "min_age_seconds": ok} + src = FileSource(Source(type=ConnectorType.FILE, settings=settings)) + assert src.min_age_seconds == (0.0 if ok == "unset" else float(str(ok))) async def test_file_source_quarantines_corrupt_archive(tmp_path: Path) -> None: diff --git a/tests/test_x12_parsing.py b/tests/test_x12_parsing.py index 539f07d389..c83cace92c 100644 --- a/tests/test_x12_parsing.py +++ b/tests/test_x12_parsing.py @@ -671,6 +671,73 @@ def test_frame_reader_crlf_terminator_byte_at_a_time() -> None: assert out == [data] +def test_frame_reader_marks_only_an_interchanges_own_line_ends_as_trailer_only() -> None: + # vault BACKLOG #2938: up to two CR/LF bytes after a completed interchange are its ending, so a + # listener's frame deadline must not start on them. Mirrors the FrameDecoder test in + # tests/test_listener_admission.py. + data = interchange().encode("utf-8") + reader = X12FrameReader() + assert list(reader.feed(data)) == [data] + assert not reader.trailer_only, "a read that completed an interchange is not trailer-only" + assert list(reader.feed(b"\r")) == [] + assert reader.trailer_only, "the CR after the IEA terminator" + assert list(reader.feed(b"\n")) == [] + assert reader.trailer_only, "the LF of a CR LF ending" + # CONTROL: a third line-end byte is past the allowance, so it is noise like any other byte. + assert list(reader.feed(b"\r")) == [] + assert not reader.trailer_only + # CONTROL: a byte that is not a line end ends the allowance for what follows it too. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"Z")) == [] + assert not reader.trailer_only + assert list(reader.feed(b"\n")) == [] + assert not reader.trailer_only + # A line end that arrived with the interchange uses up part of the allowance. + assert list(reader.feed(data + b"\r")) == [data] + assert list(reader.feed(b"\n")) == [] + assert reader.trailer_only + assert list(reader.feed(b"\n")) == [] + assert not reader.trailer_only + # Both line ends in one read of their own. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"\r\n")) == [] + assert reader.trailer_only + # An empty read is not trailer-only, and holds no byte to end the allowance with. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"")) == [] + assert not reader.trailer_only + assert list(reader.feed(b"\r")) == [] + assert reader.trailer_only + # CONTROL: three in one read are past the allowance. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"\r\n\r")) == [] + assert not reader.trailer_only + # Decided when feed() is called, before its generator runs, so it never describes a past read. + assert list(reader.feed(data)) == [data] + reader.feed(b"Z") # not iterated + assert not reader.trailer_only + # Before any interchange there is nothing for a line end to end. + fresh = X12FrameReader() + assert list(fresh.feed(b"\r\n")) == [] + assert not fresh.trailer_only + # A CR+LF segment terminator is the interchange's own, so the allowance starts after it. + crlf = interchange(term="\r\n").encode("utf-8") + assert list(fresh.feed(crlf)) == [crlf] + assert list(fresh.feed(b"\r\n")) == [] + assert fresh.trailer_only + + +def test_frame_reader_trailer_bytes_still_reach_the_noise_buffer() -> None: + # The flag only observes. The line ends still sit in the 2-byte noise tail, so the leftover + # reading the reuse path takes from the buffer, and the next interchange's framing, are unchanged. + data = interchange().encode("utf-8") + reader = X12FrameReader() + assert list(reader.feed(data + b"\r\n")) == [data] + assert reader._buf == b"\r\n" + assert list(reader.feed(data)) == [data] + assert not reader.in_frame + + def test_integrity_group_and_transaction_mismatches() -> None: # GS06/GE02 control-number mismatch. assert any("GE02" in p for p in check_integrity(interchange().replace("GE*1*1~", "GE*1*9~")))