Repository: nats-io/nats.py

Issue:
Passing wrong stream name to `add_stream` throws `errors.TimeoutError`
This may be similar to #305 and/or #437.

Following code:
```python
import nats
import pytest

@pytest.mark.asyncio
async def test_js_wrong_name():
    nc = await nats.connect("localhost")
    js = nc.jetstream()
    await js.add_stream(name="stream.with.wrong.name")

    await nc.close()
```

raises timeout on `js.add_stream`:

```
>           raise errors.TimeoutError
E           nats.errors.TimeoutError: nats: timeout

../.venv/lib/python3.11/site-packages/nats/aio/client.py:1025: TimeoutError
```

If the stream name is correct it's ok.

Discussion:

Patch:
diff --git a/nats/js/api.py b/nats/js/api.py
index 48c3836..3205b02 100644
--- a/nats/js/api.py
+++ b/nats/js/api.py
@@ -276,6 +276,7 @@ class StreamConfig(Base):
     max_msgs: Optional[int] = None
     max_bytes: Optional[int] = None
     discard: Optional[DiscardPolicy] = DiscardPolicy.OLD
+    discard_new_per_subject: bool = False
     max_age: Optional[float] = None  # in seconds
     max_msgs_per_subject: int = -1
     max_msg_size: Optional[int] = -1
diff --git a/nats/js/client.py b/nats/js/client.py
index ef76813..0ee1365 100644
--- a/nats/js/client.py
+++ b/nats/js/client.py
@@ -19,7 +19,7 @@ import json
 import time
 from email.parser import BytesParser
 from secrets import token_hex
-from typing import TYPE_CHECKING, Awaitable, Callable, Optional, List, Dict
+from typing import TYPE_CHECKING, Awaitable, Callable, Optional, List, Dict, Any
 
 import nats.errors
 import nats.js.errors
@@ -165,7 +165,7 @@ class JetStreamContext(JetStreamManager):
         payload: bytes = b'',
         timeout: Optional[float] = None,
         stream: Optional[str] = None,
-        headers: Optional[Dict] = None
+        headers: Optional[Dict[str, Any]] = None
     ) -> api.PubAck:
         """
         publish emits a new message to JetStream and waits for acknowledgement.
diff --git a/nats/js/manager.py b/nats/js/manager.py
index df7dd86..9864865 100644
--- a/nats/js/manager.py
+++ b/nats/js/manager.py
@@ -91,12 +91,26 @@ class JetStreamManager:
         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,
         )


Test patch:
diff --git a/tests/test_js.py b/tests/test_js.py
index ec61391..f6d632b 100644
--- a/tests/test_js.py
+++ b/tests/test_js.py
@@ -12,6 +12,7 @@ import json
 import io
 import tempfile
 from unittest.mock import AsyncMock
+import re
 
 import pytest
 import nats
@@ -1968,6 +1969,106 @@ class AckPolicyTest(SingleJetStreamServerTestCase):
         assert task.done()
         assert received
 
+class DiscardPolicyTest(SingleJetStreamServerTestCase):
+  @async_test
+  async def test_with_discard_new_and_discard_new_per_subject_set(self):
+      # Connect to NATS and create JetStream context
+      nc = await nats.connect()
+      js = nc.jetstream()
+
+      stream_name = "FOO0"
+      config = nats.js.api.StreamConfig(
+          name=stream_name,
+          discard=nats.js.api.DiscardPolicy.NEW,
+          discard_new_per_subject=True,
+          max_msgs_per_subject=100
+      )
+
+      await js.add_stream(config)
+      info = await js.stream_info(stream_name)
+      self.assertEqual(info.config.discard, nats.js.api.DiscardPolicy.NEW)
+      self.assertEqual(info.config.discard_new_per_subject, True)
+
+      # Close the NATS connection after the test
+      await nc.close()
+
+  @async_test
+  async def test_with_discard_new_and_discard_new_per_subject_not_set(self):
+      # Connect to NATS and create JetStream context
+      nc = await nats.connect()
+      js = nc.jetstream()
+
+      stream_name = "FOO1"
+      config = nats.js.api.StreamConfig(
+          name=stream_name,
+          discard=nats.js.api.DiscardPolicy.NEW,
+          discard_new_per_subject=False,
+          max_msgs_per_subject=100
+      )
+
+      await js.add_stream(config)
+      info = await js.stream_info(stream_name)
+      self.assertEqual(info.config.discard, nats.js.api.DiscardPolicy.NEW)
+      self.assertEqual(info.config.discard_new_per_subject, False)
+
+      await nc.close()
+
+  @async_test
+  async def test_with_discard_old_and_discard_new_per_subject_set(self):
+      # Connect to NATS and create JetStream context
+      nc = await nats.connect()
+      js = nc.jetstream()
+
+      stream_name = "FOO2"
+      config = nats.js.api.StreamConfig(
+          name=stream_name,
+          discard='DiscardOld',
+          discard_new_per_subject=True,
+          max_msgs_per_subject=100
+      )
+
+      with self.assertRaises(APIError):
+          await js.add_stream(config)
+
+      # Close the NATS connection after the test
+      await nc.close()
+
+  @async_test
+  async def test_with_discard_old_and_discard_new_per_subject_not_set(self):
+      # Connect to NATS and create JetStream context
+      nc = await nats.connect()
+      js = nc.jetstream()
+
+      stream_name = "FOO3"
+      config = nats.js.api.StreamConfig(
+          name=stream_name,
+          discard='DiscardOld',
+          discard_new_per_subject=True,
+          max_msgs_per_subject=100
+      )
+
+      with self.assertRaises(APIError):
+          await js.add_stream(config)
+
+      await nc.close()
+
+  @async_test
+  async def test_with_discard_new_and_discard_new_per_subject_set_no_max_msgs(self):
+      # Connect to NATS and create JetStream context
+      nc = await nats.connect()
+      js = nc.jetstream()
+
+      stream_name = "FOO4"
+      config = nats.js.api.StreamConfig(
+          name=stream_name,
+          discard=nats.js.api.DiscardPolicy.NEW,
+          discard_new_per_subject=True
+      )
+
+      with self.assertRaises(APIError):
+          await js.add_stream(config)
+
+      await nc.close()
 
 class OrderedConsumerTest(SingleJetStreamServerTestCase):
 
@@ -4267,3 +4368,33 @@ class V210FeaturesTest(SingleJetStreamServerTestCase):
         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)