asvs: approval compensation, upload quota atomicity, TOTP SHA-256 cutover, MLLP inbound pacing

docs/CONNECTIONS.md

@@ -286,6 +286,8 @@ duplicate name (across **any** of these files) and an inbound that binds a route
 | `max_connections` | in | `256` | cap on concurrent client connections (connection-flood guard). `None`/`0` = unlimited. |
 | `receive_timeout` | in | `60.0` | close a client idle this many seconds (slowloris guard). `None`/`0` = no timeout. |
 | `max_frame_bytes` | both | `16 MiB` | reject a single MLLP frame larger than this before buffering it whole (OOM guard); applies to inbound frames and outbound ACKs. `None`/`0` = unlimited. |
+| `max_messages_per_second` | in | **off** | sustained message-rate ceiling per **connection** (ASVS 2.4.1 / 15.2.2). Over budget the listener **pauses reading**, so TCP back-pressures the sender — **no message is ever dropped, refused or NAK'd**, and none is reordered. Unset = no bound, which is a deliberate exception to this table's usual secure-default rule: a guessed rate on a clinical interface throttles real traffic, so the number has to come from your own feed profile. |
+| `message_burst` | in | = 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 a connection can always make progress. |
 | `connect_timeout` | out | `10.0` | TCP connect timeout (s) |
 | `timeout_seconds` | out | `30.0` | wait this long for the ACK |
 | `no_ack` | out | `false` | **(BACKLOG #117, ADR 0124) fire-and-forward (MLLP outbound only):** when `true`, deliver on the successful TCP **write** and read **no** ACK — delivery is confirmed on write, **not** on a positive MSA-1 ACK, so there is **no NAK- or timeout-driven retry** (*at-most-once-confirmation*). A connect/drain failure is still charged and retried (at-least-once for the write; a retry may duplicate — receivers stay idempotent). Composes with `persistent=true` (no handshake **and** no ACK wait — the max-throughput non-acking posture). **Incompatible with `capture_response`/`reingress_to`** (nothing to capture) and MLLP-only — both rejected at `check`. `false` (default) = **byte-identical** (read + validate one ACK). |

docs/SECURITY.md

@@ -1592,7 +1592,7 @@ multi-host deployment must additionally front the API with a proxy/WAF limiter a
 | Request body | `[store].max_upload_bytes` (the `/uploads` routes only) | 1 MiB elsewhere | per request | no | no | no | **stateless** — every route, in ASGI middleware | **413** over the cap, **400** on ambiguous CL+TE framing or an invalid `Content-Length`, **411** on a chunked body |
 | OIDC pending flows | `[auth].oidc_flow_cache_max` (global), `DEFAULT_PER_IP_CAP` (per-IP, no knob), `oidc_flow_ttl_seconds` | 512 / 16 / 300 s | 300 s TTL | no | **yes** (512) | **yes** (16) | **in-process** — `GET /ui/oidc/start` — reject-when-full, never evict | 303 → `/ui/login?e=rate_limited`, WARNING-logged, **never** audited |
 | WebAuthn pending ceremonies | `GLOBAL_PENDING_CAP`, `PER_USER_PENDING_CAP`, `CHALLENGE_TTL_SECONDS` (module constants, no knobs) | 4096 / 16 / 120 s | 120 s TTL | **yes** (16) | **yes** (4096) | no | **in-process** — every passkey registration + assertion ceremony | per-user: evicts that user's **own** oldest pending ceremony (silent); global: `ChallengeCacheFullError` naming the cause + the `admin_reset_mfa` recovery path |
-| **Ingest plane** | *(none)* | — | — | — | — | — | **n/a** — inbound connections | **no message-rate or volume limit exists.** Inbound carries resource caps only — `max_connections` (256), `receive_timeout` (60.0 s), `max_frame_bytes` (16 MiB), per-connection `max_message_bytes`, `source_ip_allowlist` — and nothing in `transports/`, `config/` or `pipeline/` exposes a messages-per-second control |
+| **Ingest plane** | `max_messages_per_second`, `message_burst` (MLLP inbound) | **off** | per message | no | no | no | **in-process** — one bucket per MLLP connection, so it neither coordinates across engine shards nor aggregates per peer | **exists but ships OFF, so unset there is still no volume bound.** When set, the listener **pauses reading** over budget so TCP back-pressures the sender: no message is dropped, refused, NAK'd or reordered (the count-and-log invariant forbids accept-and-drop, so a discarding limiter was never available). Bounded by the bucket deficit. The off default is **ruled, not accidental** — a rate on a clinical interface is only safe at a number from a real feed profile. **Not covered:** the raw-TCP inbound, and any per-peer bound (MLLP peers are unauthenticated, so the only key would be source IP, which NAT collapses). Other inbound caps are resource-only — `max_connections` (256), `receive_timeout` (60.0 s), `max_frame_bytes` (16 MiB), per-connection `max_message_bytes`, `source_ip_allowlist` |
 
 **What these limits defend, and what they do not.** The full inventory of resource-demanding
 functionality — including the surfaces that remain **unbounded** at this release — is

messagefoundry/api/app.py

@@ -506,7 +506,10 @@ async def _purge(p: Mapping[str, Any]) -> dict[str, Any]:
         # Load-bearing dual-control guard (findings #1/#4/#11): ApprovalGate.approve runs THIS executor
         # directly (purge_connection is NOT re-entered on the release path), and it flips the row to
         # 'approved' BEFORE executing — so the require-quiesced precondition must be re-checked HERE, and
-        # a failure must NOT raise (a raise would strand the row approved-but-unexecuted). A non-quiesced
+        # a failure should NOT raise. (Since ASVS 2.3.3 the gate compensates a raise by rolling the row
+        # to 'failed' and auditing it, so a raise no longer strands it approved-but-unexecuted; skipping
+        # is still the better outcome HERE, because a non-quiesced outbound is a retryable precondition
+        # miss the operator can clear, not a failed operation.) A non-quiesced
         # (running/stopping) outbound could have an INFLIGHT row cancel_queued cannot cancel, so purging
         # it would mis-fire; skip fail-closed and record cancelled=0/skipped in the approval audit. The
         # operator re-Stops (lets it quiesce) and re-requests.

messagefoundry/api/approvals.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import time
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
@@ -25,6 +26,8 @@
 from messagefoundry.config.settings import ApprovalsSettings
 from messagefoundry.store.base import Store
 
+log = logging.getLogger(__name__)
+
 #: An executor re-runs a captured operation on approval, returning a small JSON-able result summary.
 Executor = Callable[[Mapping[str, Any]], Awaitable[dict[str, Any]]]
 
@@ -130,7 +133,28 @@
         ):
             raise ApprovalError(409, "request was already decided")
         params = json.loads(str(row["params"]))
-        result = await op.execute(params)
+        try:
+            result = await op.execute(params)
+        except Exception as exc:
+            # ASVS 2.3.3 COMPENSATING TRANSITION. The row moved to 'approved' BEFORE the executor ran
+            # (that ordering is load-bearing — it guards the double-approve race — and must stay). If
+            # the executor raises, the row would otherwise be stranded asserting an operation that
+            # never happened, and no approval.approved row is written either, so the store would carry
+            # an approval with no outcome at all. Roll it to 'failed' and audit the failure against
+            # both identities, then re-raise so the caller still sees the error.
+            #
+            # `except Exception` is deliberate and is not a swallow: ANY executor failure has to
+            # compensate, and the original is re-raised below. BaseException (notably CancelledError)
+            # is intentionally NOT caught — a cancelled approve must not be recorded as a failure.
+            await self._compensate_failed_execution(
+                approval_id,
+                operation=operation,
+                approver=approver,
+                requester=str(row["requester"]),
+                error=exc,
+                client=client,
+            )
+            raise
         await self._store.record_audit(
             "approval.approved",
             actor=approver,
@@ -154,6 +178,54 @@
             "result": result,
         }
 
+    async def _compensate_failed_execution(
+        self,
+        approval_id: str,
+        *,
+        operation: str,
+        approver: str,
+        requester: str,
+        error: BaseException,
+        client: str | None,
+    ) -> None:
+        """Roll a released-but-unexecuted request back out of ``approved`` (ASVS 2.3.3).
+
+        Best effort by construction: the caller re-raises the ORIGINAL executor error either way, so
+        a store that is itself unreachable here must not mask the error that actually explains the
+        failure. A compensation failure is logged loudly rather than swallowed."""
+        try:
+            # Guarded on 'approved' so this can never clobber a row another caller rejected or
+            # expired, and so a re-drive of the same failure is idempotent (second call moves 0 rows).
+            moved = await self._store.decide_pending_approval(
+                approval_id,
+                status="failed",
+                approver=approver,
+                decided_at=time.time(),
+                from_status="approved",
+            )
+            await self._store.record_audit(
+                "approval.failed",
+                actor=approver,
+                detail=json.dumps(
+                    {
+                        "approval_id": approval_id,
+                        "operation": operation,
+                        "requester": requester,
+                        # The type, never the message: an executor's exception text can carry
+                        # connection names, paths or params, and the audit log is not a PHI sink.
+                        "error": type(error).__name__,
+                        "compensated": moved,
+                    }
+                ),
+                client=client,
+            )
+        except Exception:  # noqa: BLE001 - see the docstring; the original error must win
+            log.exception(
+                "approval %s: executor failed AND the compensating transition failed; the row may "
+                "still read 'approved' for an operation that did not run",
+                approval_id,
+            )
+
     async def reject(
         self, approval_id: str, *, approver: str, client: str | None = None
     ) -> dict[str, Any]:

messagefoundry/auth/totp.py

@@ -46,7 +46,13 @@
 #: ± steps of clock skew tolerated at verify time (one step each side ≈ 30 s).
 DEFAULT_WINDOW = 1
 
-_SECRET_BYTES = 20  # 160 bits — RFC 4226 recommends ≥ 128 bits, 160 for HMAC-SHA1
+# 160 bits. RFC 4226 requires >= 128 and recommends 160. RFC 6238 R6 additionally says the key
+# SHOULD match the HMAC output length -- 32 bytes now that the digest is SHA-256 (see _TOTP_DIGEST).
+# Kept at 20 DELIBERATELY: that clause is about interoperability convention rather than strength, 160
+# bits is ample against HMAC-SHA256, and 32 bytes would lengthen manual entry from 32 to 52 base32
+# characters on a screen an operator types from. Revisit only with a real interop failure, not on the
+# SHOULD alone.
+_SECRET_BYTES = 20
 
 # Recovery codes: human-legible groups from an unambiguous alphabet (no 0/O/1/I/L confusion).
 _RECOVERY_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
@@ -66,9 +72,25 @@ def _decode_secret(secret: str) -> bytes:
     return base64.b32decode(cleaned + padding, casefold=True)
 
 
+#: The HOTP MAC. RFC 6238 permits SHA-1, SHA-256 and SHA-512; SHA-1 is the RFC default and was the
+#: shipped choice until 2026-08-11, when it was retired under the G18 ruling.
+#:
+#: THIS CONSTANT AND THE ADVERTISED ALGORITHM MUST NEVER DIVERGE. The authenticator computes with
+#: whatever ``otpauth_uri`` told it and the engine computes with this — so a change to one alone does
+#: not fail loudly, it silently produces codes that never match, for every user, with no diagnostic.
+#: That is why :data:`_TOTP_ALGORITHM` is DERIVED from this rather than written beside it: the two
+#: cannot be edited apart. ``hashlib.sha256().name.upper()`` is exactly the otpauth spelling, and the
+#: same derivation is correct for all three permitted digests.
+_TOTP_DIGEST = hashlib.sha256
+_TOTP_ALGORITHM = _TOTP_DIGEST().name.upper()
+
+
 def _hotp(key: bytes, counter: int, digits: int) -> str:
-    """RFC 4226 HOTP: HMAC-SHA1 over the 8-byte counter, dynamically truncated to ``digits`` decimals."""
-    mac = hmac.new(key, counter.to_bytes(8, "big"), hashlib.sha1).digest()
+    """RFC 4226 HOTP over the 8-byte counter, dynamically truncated to ``digits`` decimals.
+
+    The MAC is :data:`_TOTP_DIGEST` (SHA-256 since 2026-08-11), not RFC 4226's SHA-1.
+    """
+    mac = hmac.new(key, counter.to_bytes(8, "big"), _TOTP_DIGEST).digest()
     offset = mac[-1] & 0x0F
     truncated = int.from_bytes(mac[offset : offset + 4], "big") & 0x7FFFFFFF
     return str(truncated % (10**digits)).zfill(digits)
@@ -170,14 +192,23 @@ def otpauth_uri(
     period: int = DEFAULT_PERIOD,
     digits: int = DEFAULT_DIGITS,
 ) -> str:
-    """Build the ``otpauth://totp/…`` URI an authenticator app scans (the UI renders it as a QR code)."""
+    """Build the ``otpauth://totp/…`` URI an authenticator app scans (the UI renders it as a QR code).
+
+    **Advertises SHA-256, and enrolling apps must honour it.** Most modern authenticators do (1Password,
+    Bitwarden, Aegis, FreeOTP, Authy). **Google Authenticator historically IGNORES the ``algorithm``
+    parameter and computes SHA-1 regardless** — against which this engine's codes will simply never
+    match, with no error to explain why. That is the known cost of the 2026-08-11 SHA-1 retirement
+    (G18), accepted while there are zero enrolled users; it is a support burden, not a security one.
+    An operator hitting it needs an app that honours the parameter, not a re-enrolment.
+    """
     # The "issuer:account" colon is the conventional literal label separator (keep it; encode the rest).
     label = quote(f"{issuer}:{account}", safe=":")
     params = urlencode(
         {
             "secret": secret,
             "issuer": issuer,
-            "algorithm": "SHA1",
+            # DERIVED, never a literal — see _TOTP_DIGEST for why these two cannot be edited apart.
+            "algorithm": _TOTP_ALGORITHM,
             "digits": digits,
             "period": period,
         }

messagefoundry/config/settings.py

@@ -436,8 +436,12 @@ class StoreSettings(_Section):
     # PHI-at-rest is age-pruned. Enforced in `UploadStore.save` (a would-be over-quota upload is refused
     # HTTP 409 before any write, audited `upload.reject_quota`) and by an age-based prune sweep (blob+meta
     # pairs older than `uploads_retention_days` are deleted, opportunistically at save time plus a periodic
-    # task, each prune audited `upload.prune`). Quotas are per-process per-`uploads_dir` (multiple engine
-    # shards at one dir multiply the budget — a documented residual, same shape as the summary-rate cap).
+    # task, each prune audited `upload.prune`). Quotas are enforced per-`uploads_dir`, NOT per-process:
+    # the check reads the sidecars off disk with no cache, so engine shards sharing one dir see each
+    # other's files and the budget does NOT multiply (measured 2026-08-10 — two UploadStores over one
+    # dir, the second refused the same uploader at quota, against a live positive control). What IS
+    # shared across them is the check-then-write race below, which overshoots by at most one file per
+    # concurrently in-flight upload. Shards given SEPARATE dirs get separate budgets, by construction.
     max_upload_files_per_user: int = Field(
         default=100,
         ge=1,

messagefoundry/store/base.py

@@ -1486,7 +1486,13 @@ async def get_pending_approval(self, approval_id: str) -> Row | None: ...
     async def list_pending_approvals(self, *, now: float, limit: int = 100) -> Sequence[Row]: ...
 
     async def decide_pending_approval(
-        self, approval_id: str, *, status: str, approver: str | None, decided_at: float
+        self,
+        approval_id: str,
+        *,
+        status: str,
+        approver: str | None,
+        decided_at: float,
+        from_status: str = "pending",
     ) -> bool: ...
 
     async def audit_anchor(self) -> tuple[int, str]: ...

messagefoundry/store/postgres.py

@@ -5985,17 +5985,25 @@ async def list_pending_approvals(self, *, now: float, limit: int = 100) -> Seque
         )
 
     async def decide_pending_approval(
-        self, approval_id: str, *, status: str, approver: str | None, decided_at: float
+        self,
+        approval_id: str,
+        *,
+        status: str,
+        approver: str | None,
+        decided_at: float,
+        from_status: str = "pending",
     ) -> bool:
-        """Atomically move a still-``pending`` request to ``status`` (approved/rejected/expired).
-        Returns ``True`` iff this call made the transition — guards against a double decision."""
+        """Atomically move a request in ``from_status`` to ``status``.
+        Returns ``True`` iff this call made the transition — guards against a double decision.
+        The SQLite twin documents why the guard is a parameter (ASVS 2.3.3)."""
         result = await self._pool.execute(
             "UPDATE pending_approvals SET status = $1, approver = $2, decided_at = $3"
-            " WHERE id = $4 AND status = 'pending'",
+            " WHERE id = $4 AND status = $5",
             status,
             approver,
             decided_at,
             approval_id,
+            from_status,
         )
         return _rowcount(result) > 0
 

messagefoundry/store/sqlserver.py

@@ -9015,16 +9015,23 @@ async def list_pending_approvals(self, *, now: float, limit: int = 100) -> list[
         )
 
     async def decide_pending_approval(
-        self, approval_id: str, *, status: str, approver: str | None, decided_at: float
+        self,
+        approval_id: str,
+        *,
+        status: str,
+        approver: str | None,
+        decided_at: float,
+        from_status: str = "pending",
     ) -> bool:
-        """Atomically move a still-``pending`` request to ``status`` (approved/rejected/expired).
-        Returns ``True`` iff this call made the transition — guards against a double decision."""
+        """Atomically move a request in ``from_status`` to ``status``.
+        Returns ``True`` iff this call made the transition — guards against a double decision.
+        The SQLite twin documents why the guard is a parameter (ASVS 2.3.3)."""
         async with self._acquire() as conn, self._cursor(conn) as cur:
             try:
                 await cur.execute(
                     "UPDATE pending_approvals SET status = ?, approver = ?, decided_at = ?"
-                    " WHERE id = ? AND status = 'pending'",
-                    (status, approver, decided_at, approval_id),
+                    " WHERE id = ? AND status = ?",
+                    (status, approver, decided_at, approval_id, from_status),
                 )
                 count = cur.rowcount
                 await self._commit(conn)