From 41dff01497ebf8a95643fa64cb04e99db41adf2c Mon Sep 17 00:00:00 2001 From: Thorsten Date: Sat, 18 Jul 2026 13:40:11 +0200 Subject: [PATCH] Fix NATS classifier publish dropping messages with quoted multi-word args A shlex-quoted plugin argument (e.g. choose 'multi word option') produces a routing key containing a literal space. AMQP's topic exchange tolerates that, but NATS's whitespace-delimited wire protocol does not - the PUB frame gets corrupted and the server silently drops the connection before the message reaches any subscriber. Sanitize whitespace out of the subject before publishing; the JSON payload plugins actually read from is untouched. Co-Authored-By: Claude Sonnet 5 --- src/distbot/common/broker_nats.py | 19 ++++++++++++++++++- tests/test_unit/test_broker_nats.py | 20 ++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) create mode 100644 tests/test_unit/test_broker_nats.py diff --git a/src/distbot/common/broker_nats.py b/src/distbot/common/broker_nats.py index 824e71d..cf37b67 100644 --- a/src/distbot/common/broker_nats.py +++ b/src/distbot/common/broker_nats.py @@ -25,6 +25,7 @@ Design notes (see the migration plan for the full rationale): """ import asyncio import logging +import re import threading from typing import Callable, List @@ -48,6 +49,22 @@ def _to_bytes(value) -> bytes: return value.encode("utf-8") if isinstance(value, str) else value +_WHITESPACE = re.compile(r"\s+") + + +def _sanitize_subject(subject: str) -> str: + # Routing keys come from shlex-quoted user text (e.g. "choose 'multi + # word option'") and can legally contain spaces under AMQP's topic + # exchange, which treats a routing key as an arbitrary string. NATS's + # wire protocol is whitespace-delimited ("PUB <#bytes>"), so a + # literal space inside a subject corrupts the PUB frame and the server + # drops the connection - confirmed live, message never reaches any + # subscriber. Since '#' is translated to ".*" by routing.matches() and + # absorbs any character here, substituting whitespace doesn't affect + # dispatch matching. + return _WHITESPACE.sub("_", subject) + + def _stream_name_for(queue: str) -> str: return queue.upper() @@ -152,7 +169,7 @@ class NatsSyncBroker(SyncBroker): self._loop.close() def publish(self, subject: str, body: bytes) -> None: - full_subject = CLASSIFIER_PREFIX + _to_str(subject) + full_subject = CLASSIFIER_PREFIX + _sanitize_subject(_to_str(subject)) self._call(self._nc.publish(full_subject, _to_bytes(body))) def publish_queue(self, queue: str, body: bytes, durable: bool = True, max_age_seconds: int = None) -> None: diff --git a/tests/test_unit/test_broker_nats.py b/tests/test_unit/test_broker_nats.py new file mode 100644 index 0000000..18cd7ab --- /dev/null +++ b/tests/test_unit/test_broker_nats.py @@ -0,0 +1,20 @@ +# -*- coding: utf-8 -*- +import pytest + +from distbot.common.broker_nats import _sanitize_subject + + +@pytest.mark.parametrize("subject,expected", [ + # A quoted, multi-word plugin argument (e.g. choose "'multi word + # option'") shows up as a single shlex token containing spaces, which + # AMQP's topic exchange tolerates as an arbitrary routing key but NATS's + # whitespace-delimited wire protocol ("PUB <#bytes>") does + # not - an unsanitized space here corrupts the PUB frame and the + # message never reaches any subscriber (confirmed against a live + # nats-server). + ("nick.choose.will ich nicht", "nick.choose.will_ich_nicht"), + ("nick.choose.a\tb", "nick.choose.a_b"), + ("nick.dice.5", "nick.dice.5"), +]) +def test_sanitize_subject_strips_whitespace(subject, expected): + assert _sanitize_subject(subject) == expected -- 2.47.3