emrg: a session id no directory can carry is refused by name

emrg/server/daemon.py

@@ -103,10 +103,10 @@
 )
 from emrg.session import (
     Session,
-    is_safe_session_id,
     last_n_messages,
     records_to_messages,
     session_dir,
+    session_id_refusal,
 )
 
 
@@ -2863,10 +2863,18 @@ def _collect_history_data(self, session: Session) -> dict[str, str]:
     async def _refuse_an_unsafe_session_id(
         self, ws, msg_type: str, session_id: object
     ) -> bool:
-        """Refuse a client-supplied session id that is not one path component.
+        """Refuse a client-supplied session id that cannot name one entry of the sessions
+        directory.
 
         True means the id was refused and the client has been told, so the caller stops.
 
+        Two reasons reach this door, and it reports the one `session_id_refusal` names: an
+        id that is more than one path component (or not a name at all), and an id no
+        directory can carry (a reserved device name, or a name ending in a space or a dot
+        — measured on Windows Server 2022, where `con` reached the mkdir and came back as
+        `WinError 267`). A device name *is* one path component, so it must not be told the
+        first sentence.
+
         A session id arrives in a client message and lands on a path in half the handlers
         below: `_get_or_create_session` (eleven callers) writes it, the resume reader and
         the two memory readers open it, and `delete_session` hands the resolved path to
@@ -2881,15 +2889,15 @@ async def _refuse_an_unsafe_session_id(
         client nothing. Both call sites ask this method, so the rule has one writer even
         though it has two doors.
         """
-        if not session_id or is_safe_session_id(session_id):
+        refusal = session_id_refusal(session_id)
+        if not session_id or refusal is None:
             return False
         logger.warning(
-            "refused a %r message carrying session id %r: a session id is one "
-            "path component inside <cwd>/.emrg/sessions", msg_type, session_id,
+            "refused a %r message carrying session id %r: %s inside "
+            "<cwd>/.emrg/sessions", msg_type, session_id, refusal,
         )
         await self._send(ws, {
-            "error": f"invalid session id: {session_id!r} (a session id is "
-                     f"one path component)",
+            "error": f"invalid session id: {session_id!r} ({refusal})",
         })
         return True
 

emrg/session.py

@@ -316,8 +316,36 @@ def _refuse_a_non_regular_subject(path: Path) -> None:
 _NAME_SEPARATORS = ("/", "\\")
 
 
-def is_safe_session_id(session_id: object) -> bool:
-    """True when `session_id` names exactly one entry inside the sessions directory.
+#: The names Windows reserves for devices, in the set its own documentation gives. A
+#: directory cannot be called one of them, with or without an extension, so an id that is
+#: one of these names nothing on the platform that can receive it. The guard is the set
+#: itself and not a prefix: `console` and `con-artist` are ordinary names, and `COM0` /
+#: `LPT0` are **not** reserved — a predicate that refused them would cost the host names
+#: the platform accepts.
+_RESERVED_DEVICE_NAMES = frozenset(
+    {"con", "prn", "aux", "nul"}
+    | {f"com{i}" for i in range(1, 10)}
+    | {f"lpt{i}" for i in range(1, 10)}
+)
+
+#: Why an id is not one entry of the sessions directory, in the two sentences the two
+#: refusals use. The first is about shape — the id is more than one component, or not a
+#: name at all. The second is about the filesystem, and it needs a sentence of its own: a
+#: reserved device name **is** one path component, so the first sentence would be false
+#: about it.
+_REFUSAL_NOT_ONE_COMPONENT = "a session id is one path component"
+_REFUSAL_NOT_A_DIRECTORY_NAME = (
+    "no directory can carry this name: it is a reserved device name, or it ends in a "
+    "space or a dot, which the filesystem folds away"
+)
+
+
+def session_id_refusal(session_id: object) -> str | None:
+    """Why `session_id` cannot name one entry of the sessions directory, or `None`.
+
+    The only writer of the verdict and of the sentence beside it: `is_safe_session_id` is
+    this returning `None`, and the daemon passes the sentence to the client rather than
+    restating the rule a third time.
 
     A session id arrives in a client message, so it is an input like any other:
     `Path(cwd) / ".emrg" / "sessions" / session_id` hands `..` and an absolute id to the
@@ -330,16 +358,48 @@ def is_safe_session_id(session_id: object) -> bool:
     separator, and no drive. The drive is asked of **both** flavours — `os.path.splitdrive`
     answers `("", "C:evil")` on this host and `ntpath.splitdrive` answers `("C:", "evil")`,
     and a `C:`-relative id is a path, not a name, on the platform that can receive it.
+
+    Two further refusals keep the name one a directory can actually carry, measured on
+    Windows Server 2022 on master `6ddb7cc`: `is_safe_session_id("con")` answered `True`,
+    and `Session.create_with_id("con", <agent>)` then raised `NotADirectoryError [WinError
+    267]` out of the mkdir — an OS error about a path, for a name the host chose; `nul`
+    failed one step later, on the `memory` subdirectory (`FileNotFoundError [WinError
+    3]`); and an id ending in a space or a dot is *folded* by Windows rather than refused,
+    so it names a different directory than the id says. Both are refused here on the same
+    ground as the separator: an id reaches a path on whichever platform runs it.
     """
     if not isinstance(session_id, str) or not session_id:
-        return False
+        return _REFUSAL_NOT_ONE_COMPONENT
     if "\0" in session_id or session_id in (".", ".."):
-        return False
+        return _REFUSAL_NOT_ONE_COMPONENT
     if any(sep in session_id for sep in _NAME_SEPARATORS):
-        return False
+        return _REFUSAL_NOT_ONE_COMPONENT
     if os.path.splitdrive(session_id)[0] or ntpath.splitdrive(session_id)[0]:
-        return False
-    return True
+        return _REFUSAL_NOT_ONE_COMPONENT
+    if session_id != session_id.rstrip(" ."):
+        return _REFUSAL_NOT_A_DIRECTORY_NAME
+    if session_id.split(".")[0].strip().lower() in _RESERVED_DEVICE_NAMES:
+        return _REFUSAL_NOT_A_DIRECTORY_NAME
+    return None
+
+
+def is_safe_session_id(session_id: object) -> bool:
+    """True when `session_id` names exactly one entry inside the sessions directory.
+
+    The rule, its two sentences and the readings behind them live in
+    `session_id_refusal`, which is the one writer of the verdict; this is the form the
+    callers that only need a yes or a no ask.
+    """
+    return session_id_refusal(session_id) is None
+
+
+class UnsafeSessionId(ValueError):
+    """A session id that cannot name one entry inside the sessions directory.
+
+    Raised where a directory is about to be made from the id, so the caller learns that
+    the *name* is the problem, instead of receiving the filesystem's error about the path
+    the name turned into.
+    """
 
 
 def session_dir(cwd: Path, session_id: str) -> Path:
@@ -517,7 +577,17 @@ def create(cls, cwd: Path) -> Session:
 
     @classmethod
     def create_with_id(cls, session_id: str, cwd: Path) -> Session:
-        """Create a new session with a specific session ID (from client)."""
+        """Create a new session with a specific session ID (from client).
+
+        The id is asked first, because this is the factory that makes a directory out of
+        it: a name no directory can carry — a reserved device name, or one ending in a
+        space or a dot — would otherwise fail inside `mkdir`, and the caller would get an
+        OS error about a path rather than a refusal about the name (measured on Windows
+        Server 2022: `NotADirectoryError [WinError 267]` for `con`).
+        """
+        refusal = session_id_refusal(session_id)
+        if refusal is not None:
+            raise UnsafeSessionId(f"refusing session id {session_id!r}: {refusal}")
         session = cls(session_id, cwd)
         now = datetime.now().isoformat()
         session._created_at = now

tests/test_session_id_names_a_directory.py

@@ -0,0 +1,99 @@
+"""A session id becomes a directory, so a name no directory can carry is refused here.
+
+Measured on Windows Server 2022, master `6ddb7cc`: `is_safe_session_id("con")` answered
+`True`, and `Session.create_with_id("con", <agent>)` then raised `NotADirectoryError
+[WinError 267]` out of the mkdir — an operating-system error about a path, for a name the
+host chose. `nul` got one step further and failed on the `memory` sub-directory
+(`FileNotFoundError [WinError 3]`), and an id ending in a space or a dot is *folded* by
+Windows rather than refused, so it names a different directory than the id says.
+
+The accepted half is the discriminating one: `console`, `con-artist`, `com10` and `lpt0`
+sit beside the reserved set without being in it, and a predicate that refused them would
+cost the host names the platform accepts. Nothing here bans a space or a non-ASCII name —
+`My Task` and `中文任务` are ordinary session names and stay so.
+"""
+
+import pytest
+
+from emrg.session import (
+    Session,
+    UnsafeSessionId,
+    is_safe_session_id,
+    session_id_refusal,
+)
+
+
+@pytest.mark.parametrize("session_id", [
+    "con",
+    "CON",
+    "Con",
+    "con.txt",
+    "nul",
+    "nul.tar.gz",
+    "prn",
+    "aux",
+    "com1",
+    "com9",
+    "lpt1",
+    "lpt9",
+])
+def test_a_reserved_device_name_is_refused(session_id):
+    """Windows cannot make a directory of these, with or without an extension."""
+    assert is_safe_session_id(session_id) is False
+    assert session_id_refusal(session_id) is not None
+
+
+@pytest.mark.parametrize("session_id", [
+    "com10",     # not in the documented set: the set stops at COM9
+    "lpt0",      # ... and at LPT9; LPT0 is not reserved
+    "com0",
+    "console",   # a prefix is not a name
+    "con-artist",
+    "con_task",
+    "my con",
+    "cont",
+])
+def test_a_name_beside_the_reserved_set_is_accepted(session_id):
+    """The other direction: refusing a name the platform accepts costs the host a name."""
+    assert is_safe_session_id(session_id) is True
+    assert session_id_refusal(session_id) is None
+
+
+@pytest.mark.parametrize("session_id", ["a ", "a.", "sess ", "sess.", "con ", "con."])
+def test_a_name_ending_in_a_space_or_a_dot_is_refused(session_id):
+    """Windows strips them, so the directory is not the one the id names."""
+    assert is_safe_session_id(session_id) is False
+
+
+@pytest.mark.parametrize("session_id", ["My Task", "\u4e2d\u6587\u4efb\u52a1", "sess-1.2", "a.b"])
+def test_an_ordinary_name_is_accepted(session_id):
+    """A space and a non-ASCII name are session names, not defects — and a dot inside a
+    name is fine, only a trailing one is folded away."""
+    assert is_safe_session_id(session_id) is True
+
+
+def test_creating_a_session_with_a_device_name_is_refused_by_name(tmp_path):
+    """The refusal names the rule, and no directory is attempted."""
+    with pytest.raises(UnsafeSessionId) as refusal:
+        Session.create_with_id("con", tmp_path)
+
+    assert "reserved device name" in str(refusal.value)
+    assert not (tmp_path / ".emrg" / "sessions" / "con").exists()
+    assert not (tmp_path / ".emrg" / "sessions").exists()
+
+
+def test_creating_a_session_with_an_ordinary_name_still_works(tmp_path):
+    """The control: the refusal must not cost the feature it guards."""
+    session = Session.create_with_id("My Task", tmp_path)
+    assert session.session_id == "My Task"
+    assert (tmp_path / ".emrg" / "sessions" / "My Task" / "meta.json").is_file()
+
+
+def test_the_two_refusals_are_two_sentences():
+    """A device name *is* one path component, so the shape sentence would be false."""
+    shape = session_id_refusal("../../../evil")
+    device = session_id_refusal("con")
+
+    assert shape == "a session id is one path component"
+    assert device is not None and device != shape
+    assert "device" in device