Validate stream name in add_stream method (#607)

Signed-off-by: ff137 <[email protected]>

nats/js/manager.py

@@ -91,12 +91,26 @@ async def add_stream(
         if config is None:
             config = api.StreamConfig()
         config = config.evolve(**params)
-        if config.name is None:
+
+        stream_name = config.name
+        if stream_name is None:
             raise ValueError("nats: stream name is required")
 
+        # Validate stream name
+        invalid_chars = set(".*>/\\")
+        has_invalid_chars = any(char in stream_name for char in invalid_chars)
+        has_whitespace = any(char.isspace() for char in stream_name)
+        is_not_printable = not stream_name.isprintable()
+
+        if has_invalid_chars or has_whitespace or is_not_printable:
+            raise ValueError(
+                f"nats: stream name ({stream_name}) is invalid. Names cannot contain whitespace, '.', '*', '>', "
+                "path separators (forward or backward slash), or non-printable characters."
+            )
+
         data = json.dumps(config.as_dict())
         resp = await self._api_request(
-            f"{self._prefix}.STREAM.CREATE.{config.name}",
+            f"{self._prefix}.STREAM.CREATE.{stream_name}",
             data.encode(),
             timeout=self._timeout,
         )

tests/test_js.py

@@ -12,6 +12,7 @@
 import io
 import tempfile
 from unittest.mock import AsyncMock
+import re
 
 import pytest
 import nats
@@ -4367,3 +4368,33 @@ async def test_fetch_pull_subscribe_bind(self):
         assert str(err.value) == 'nats: stream name is required'
 
         await nc.close()
+
+
+class BadStreamNamesTest(SingleJetStreamServerTestCase):
+    @async_test
+    async def test_add_stream_invalid_names(self):
+        nc = NATS()
+        await nc.connect()
+        js = nc.jetstream()
+
+        invalid_names = [
+            "stream name with spaces",
+            "stream.name.with.dots",
+            "stream*name*with*asterisks",
+            "stream>name>with>greaterthans",
+            "stream/name/with/forwardslashes",
+            "stream\\name\\with\\backslashes",
+            "stream\nname\nwith\nnewlines",
+            "stream\tname\twith\ttabs",
+            "stream\x00name\x00with\x00nulls",
+        ]
+
+        for name in invalid_names:
+            with pytest.raises(
+                ValueError,
+                match=(
+                    f"nats: stream name \\({re.escape(name)}\\) is invalid. Names cannot contain whitespace, '\\.', "
+                    "'\\*', '>', path separators \\(forward or backward slash\\), or non-printable characters."
+                ),
+            ):
+                await js.add_stream(name=name)