Merge pull request #21 from snegov/feature/no-shell-and-path-injection

Stop server-provided values from reaching a shell or escaping /dsc/seafile

.github/workflows/docker-publish.yml

@@ -16,6 +16,25 @@ env:
   IMAGE: snegov/seafile-client
 
 jobs:
+  # Application failures must be distinguishable from packaging failures, so
+  # the unit tests do not depend on the image build.
+  unit-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+          cache: pip
+
+      - name: Install dependencies
+        run: pip install -r requirements-dev.txt
+
+      - name: Run unit tests
+        run: pytest -q
+
   test:
     runs-on: ubuntu-latest
 
@@ -34,7 +53,7 @@ jobs:
           tags: ${{ env.IMAGE }}:ci
 
   push:
-    needs: test
+    needs: [unit-test, test]
 
     runs-on: ubuntu-latest
     if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/')

IMPROVEMENTS.md

@@ -37,15 +37,19 @@ Acceptance criteria:
 
 ### Remove shell and path injection risks
 
-- [ ] Pass subprocess arguments as arrays instead of joining values into shell
-  command strings in `dsc/client.py` and `dsc/misc.py`.
-- [ ] Replace `su -c` with a mechanism such as `runuser --` that does not need a
-  shell.
-- [ ] Treat server-provided library names as untrusted input.
-- [ ] Resolve every target directory and prove that it remains under
-  `/dsc/seafile` before creating or writing it.
-- [ ] Define how duplicate, absolute, empty, and traversal-style library names
-  are handled.
+- [x] Pass subprocess arguments as arrays instead of joining values into shell
+  command strings in `dsc/client.py` and `dsc/misc.py`. The password is no
+  longer wrapped in quotes to survive a shell, which also fixes passwords
+  containing quotes, `$`, and backticks.
+- [x] Replace `su -c` with a mechanism such as `runuser --` that does not need a
+  shell. `runuser -u` sets the same `HOME` and a `PATH` that still finds the
+  `seaf-cli` wrapper, so the daemon environment is unchanged.
+- [x] Treat server-provided library names as untrusted input.
+- [x] Resolve every target directory and prove that it remains under
+  `/dsc/seafile` before creating or writing it. Symlinks are resolved first,
+  so an existing symlink in the libraries directory cannot redirect a write.
+- [x] Define how duplicate, absolute, empty, and traversal-style library names
+  are handled. See `dsc/paths.py` and the "Some notes" section of the README.
 
 Acceptance criteria:
 
@@ -88,6 +92,8 @@ Acceptance criteria:
 
 - [ ] Check every important subprocess return code.
 - [ ] Add timeouts to daemon startup, shutdown, and readiness polling.
+  Confirmed: `start_daemon` and `stop_daemon` poll `while True` with no bound,
+  so a daemon that never changes state hangs the container silently.
 - [ ] Handle `SIGTERM` and `SIGINT` and stop the daemon in `finally`.
 - [ ] Return a nonzero container exit code on configuration or daemon failure.
 - [ ] Produce actionable errors instead of waiting forever.
@@ -101,13 +107,16 @@ Acceptance criteria:
 ### Add real tests
 
 - [ ] Add unit tests for configuration parsing, command construction, library
-  resolution, path validation, retries, and status parsing.
+  resolution, path validation, retries, and status parsing. Command
+  construction and path validation are covered in `tests/`; configuration
+  parsing, retries, and status parsing are not.
 - [ ] Add regression tests for externally reported password and deletion
   threshold issues.
 - [ ] Add a container smoke test for imports, installed CLI version, startup,
   and shutdown using controlled fakes.
-- [ ] Keep image construction as a separate CI job so packaging failures are
-  distinguishable from application failures.
+- [x] Keep image construction as a separate CI job so packaging failures are
+  distinguishable from application failures. The `unit-test` job does not
+  depend on the image build.
 
 Acceptance criteria:
 

README.md

@@ -58,6 +58,14 @@ volumes:
 Libraries will be synced in subdirectories of `/dsc/seafile` directory inside
  container. You can mount it to host directory to access files.
 
+The subdirectory is named after the library, with whitespace replaced by
+ underscores. A library is skipped with a warning when its name cannot be a
+ directory of its own: an empty name, `.`, `..`, or a name containing `/`. Two
+ libraries whose names would produce the same directory both get their library
+ ID appended, so they never share one. Everything is created under the
+ libraries directory (`/dsc/seafile`, or the deprecated `/data` when that
+ exists); a library name can never place it elsewhere.
+
 `hostname` parameter is optional, but it's recommended to set it to some unique
  value, it will be shown in Seafile webUI as client name (`terminal-dsc` in
  given example).

dsc/client.py

@@ -12,7 +12,7 @@ from urllib3.util.retry import Retry
 import seafile
 
 from dsc import const
-from dsc.misc import create_dir, hide_password
+from dsc.misc import create_dir, hide_password, user_cmd
 
 _lg = logging.getLogger(__name__)
 
@@ -56,8 +56,8 @@ class SeafileClient:
     def __str__(self):
         return f"SeafileClient({self.user}@{self.url})"
 
-    def __gen_cmd(self, cmd: str) -> list:
-        return ["su", "-", const.DEFAULT_USERNAME, "-c", cmd]
+    def __gen_cmd(self, argv: list) -> list:
+        return user_cmd(argv)
 
     @property
     def token(self):
@@ -87,8 +87,8 @@ class SeafileClient:
 
     @property
     def daemon_ready(self) -> bool:
-        cmd = "seaf-cli status"
-        _lg.info("Checking seafile daemon status: %s", cmd)
+        cmd = ["seaf-cli", "status"]
+        _lg.info("Checking seafile daemon status: %s", " ".join(cmd))
         proc = subprocess.run(
             self.__gen_cmd(cmd),
             stdout=subprocess.DEVNULL,
@@ -99,13 +99,13 @@ class SeafileClient:
     def init_config(self):
         if self.config_initialized:
             return
-        cmd = "seaf-cli init -d %s" % self.app_dir
-        _lg.info("Initializing seafile config: %s", cmd)
+        cmd = ["seaf-cli", "init", "-d", self.app_dir]
+        _lg.info("Initializing seafile config: %s", " ".join(cmd))
         subprocess.run(self.__gen_cmd(cmd))
 
     def start_daemon(self):
-        cmd = "seaf-cli start"
-        _lg.info("Starting seafile daemon: %s", cmd)
+        cmd = ["seaf-cli", "start"]
+        _lg.info("Starting seafile daemon: %s", " ".join(cmd))
         subprocess.run(self.__gen_cmd(cmd))
         _lg.info("Waiting for seafile daemon to start")
 
@@ -117,8 +117,8 @@ class SeafileClient:
         _lg.info("Seafile daemon is ready")
 
     def stop_daemon(self):
-        cmd = "seaf-cli stop"
-        _lg.info("Stopping seafile daemon: %s", cmd)
+        cmd = ["seaf-cli", "stop"]
+        _lg.info("Stopping seafile daemon: %s", " ".join(cmd))
         subprocess.run(self.__gen_cmd(cmd))
         _lg.info("Waiting for seafile daemon to stop")
 
@@ -135,9 +135,12 @@ class SeafileClient:
                 return lib_id
         return None
 
-    def sync_lib(self, lib_id: str, parent_dir: str = const.DEFAULT_LIBS_DIR):
-        lib_name = self.remote_libraries[lib_id]
-        lib_dir = os.path.join(parent_dir, lib_name.replace(" ", "_"))
+    def sync_lib(self, lib_id: str, lib_dir: str):
+        """
+        Sync a library into an already resolved directory. The caller decides
+        the directory, see dsc.paths.plan_lib_dirs: the name comes from the
+        server and must not be turned into a path here.
+        """
         create_dir(lib_dir)
         cmd = [
             "seaf-cli",
@@ -146,13 +149,13 @@ class SeafileClient:
             "-s", self.url,
             "-d", lib_dir,
             "-u", self.user,
-            "-p", '"' + self.password + '"',
+            "-p", self.password,
         ]
         _lg.info(
-            "Syncing library %s: %s", lib_name,
+            "Syncing library %s into %s: %s", lib_id, lib_dir,
             " ".join(hide_password(cmd, self.password)),
         )
-        subprocess.run(self.__gen_cmd(" ".join(cmd)))
+        subprocess.run(self.__gen_cmd(cmd))
 
     def __print_tx_task(self, tx_task) -> str:
         """ Print transfer task status """
@@ -234,8 +237,8 @@ class SeafileClient:
                 prev_status[library] = cur_status[library]
 
     def get_local_libraries(self) -> set:
-        cmd = "seaf-cli list"
-        _lg.info("Listing local libraries: %s", cmd)
+        cmd = ["seaf-cli", "list"]
+        _lg.info("Listing local libraries: %s", " ".join(cmd))
         out = subprocess.check_output(self.__gen_cmd(cmd))
         out = out.decode().splitlines()[1:]  # first line is a header
 
@@ -256,8 +259,8 @@ class SeafileClient:
                 continue
 
             # check current value
-            cmd = f"seaf-cli config -k {key}"
-            _lg.info("Checking seafile client option: %s", cmd)
+            cmd = ["seaf-cli", "config", "-k", key]
+            _lg.info("Checking seafile client option: %s", " ".join(cmd))
             proc = subprocess.run(self.__gen_cmd(cmd), stdout=subprocess.PIPE)
             # stdout looks like "option = value"
             cur_value = proc.stdout.decode().strip()
@@ -269,8 +272,8 @@ class SeafileClient:
                 continue
 
             # set new value
-            cmd = f"seaf-cli config -k {key} -v {value}"
-            _lg.info("Setting seafile client option: %s", cmd)
+            cmd = ["seaf-cli", "config", "-k", key, "-v", str(value)]
+            _lg.info("Setting seafile client option: %s", " ".join(cmd))
             subprocess.run(self.__gen_cmd(cmd))
             need_restart = True
 

dsc/misc.py

@@ -40,6 +40,21 @@ def create_dir(dir_path: str):
             raise RuntimeError(f"Data dir {dir_path} is not a directory")
 
 
+def user_cmd(argv: list) -> list:
+    """
+    Wrap a command so that it runs as the seafile user.
+
+    ``runuser --`` execs the command directly. ``su -c`` would hand it to a
+    login shell instead, which turns every server-provided value in the
+    command into shell code. Like ``su``, ``runuser`` sets HOME, USER and a
+    default PATH for the target user, so seaf-cli still finds its config and
+    its own wrapper.
+    """
+    if not isinstance(argv, list):
+        raise TypeError(f"Command must be a list of arguments, got {type(argv).__name__}")
+    return ["runuser", "-u", DEFAULT_USERNAME, "--"] + argv
+
+
 def hide_password(cmd: list, password: str) -> list:
     cmd = cmd.copy()
     for i, arg in enumerate(cmd):

dsc/paths.py

@@ -0,0 +1,101 @@
+"""Turn server-provided library names into safe local directories.
+
+Library names come from the Seafile server and are untrusted: a name is chosen
+by whoever can create a library, and it must never decide where the container
+writes. Every name is reduced to a single path component and every resulting
+path is proven to stay under the libraries directory.
+"""
+
+import logging
+import os
+from collections import Counter
+
+_lg = logging.getLogger(__name__)
+
+# Whitespace is folded into underscores so that library names survive as
+# readable directory names, as they always have.
+_WHITESPACE_REPLACEMENT = "_"
+
+# Names that are not usable as a directory of their own.
+_RESERVED_NAMES = {"", ".", "..", "..."}
+
+# Length of the library id used to break ties between duplicate names.
+_ID_SUFFIX_LEN = 8
+
+
+class UnsafeLibraryName(ValueError):
+    """A library name cannot be mapped to a directory under the parent."""
+
+
+def lib_dirname(lib_name: str) -> str:
+    """
+    Return the directory name for a library, or raise for a name that cannot
+    be one. Shell metacharacters are ordinary filename characters here: nothing
+    reaches a shell any more, so they are kept rather than mangled.
+    """
+    if not isinstance(lib_name, str):
+        raise UnsafeLibraryName(f"Library name is not a string: {lib_name!r}")
+
+    name = "".join(
+        _WHITESPACE_REPLACEMENT if ch.isspace() else ch
+        for ch in lib_name.strip()
+    )
+
+    if "\0" in name:
+        raise UnsafeLibraryName("Library name contains a null byte")
+    if os.sep in name or (os.altsep and os.altsep in name):
+        raise UnsafeLibraryName(f"Library name contains a path separator: {lib_name!r}")
+    if name in _RESERVED_NAMES:
+        raise UnsafeLibraryName(f"Library name is not usable as a directory: {lib_name!r}")
+
+    return name
+
+
+def resolve_lib_dir(parent_dir: str, dir_name: str) -> str:
+    """
+    Resolve ``dir_name`` inside ``parent_dir`` and prove that the result stays
+    there. Symlinks are resolved first, so an existing symlink inside the
+    parent cannot be used to write outside it.
+    """
+    parent = os.path.realpath(parent_dir)
+    resolved = os.path.realpath(os.path.join(parent, dir_name))
+
+    if resolved != parent and not resolved.startswith(parent + os.sep):
+        raise UnsafeLibraryName(
+            f"Library directory {resolved} is outside {parent}"
+        )
+    if resolved == parent:
+        raise UnsafeLibraryName(f"Library directory resolves to {parent} itself")
+
+    return resolved
+
+
+def plan_lib_dirs(libraries: dict, parent_dir: str) -> dict:
+    """
+    Map library id to its directory for every library that can safely have one.
+
+    Libraries whose names collide after whitespace folding would otherwise
+    share a directory and corrupt each other, so each of them is suffixed with
+    the start of its own id. Unsafe names are skipped with a warning rather
+    than aborting the sync of the remaining libraries.
+    """
+    names = dict()
+    for lib_id, lib_name in libraries.items():
+        try:
+            names[lib_id] = lib_dirname(lib_name)
+        except UnsafeLibraryName as err:
+            _lg.warning("Skipping library %s: %s", lib_id, err)
+
+    counts = Counter(names.values())
+    duplicates = {name for name, count in counts.items() if count > 1}
+
+    plan = dict()
+    for lib_id, name in names.items():
+        if name in duplicates:
+            name = f"{name}-{lib_id[:_ID_SUFFIX_LEN]}"
+        try:
+            plan[lib_id] = resolve_lib_dir(parent_dir, name)
+        except UnsafeLibraryName as err:
+            _lg.warning("Skipping library %s: %s", lib_id, err)
+
+    return plan

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+# The application is imported from the repository root, not installed.
+pythonpath = .

requirements-dev.txt

@@ -0,0 +1,3 @@
+-r requirements.txt
+
+pytest==9.1.1

start.py

@@ -7,6 +7,7 @@ import sys
 
 from dsc import SeafileClient, const
 from dsc.misc import setup_uid, create_dir
+from dsc.paths import plan_lib_dirs
 
 _lg = logging.getLogger('dsc')
 
@@ -76,8 +77,14 @@ def main():
     else:
         libs_dir = const.DEFAULT_LIBS_DIR
 
-    for lib_id in libs_to_sync:
-        client.sync_lib(lib_id, libs_dir)
+    # Library names come from the server, so the directory for each one is
+    # resolved and checked before anything is created or written.
+    lib_dirs = plan_lib_dirs(
+        {lib_id: client.remote_libraries[lib_id] for lib_id in libs_to_sync},
+        libs_dir,
+    )
+    for lib_id, lib_dir in lib_dirs.items():
+        client.sync_lib(lib_id, lib_dir)
     client.watch_status()
 
     client.stop_daemon()

tests/conftest.py

@@ -0,0 +1,21 @@
+"""Make ``dsc.client`` importable outside the container.
+
+``seafile`` and ``pysearpc`` ship with the Seafile CLI AppImage and only exist
+inside the image. The application tests must not need a live server or that
+AppImage, so a stub module stands in for the RPC client.
+"""
+
+import sys
+import types
+from unittest import mock
+
+
+def _install_seafile_stub():
+    if "seafile" in sys.modules:
+        return
+    seafile = types.ModuleType("seafile")
+    seafile.RpcClient = mock.MagicMock(name="RpcClient")
+    sys.modules["seafile"] = seafile
+
+
+_install_seafile_stub()

tests/test_commands.py

@@ -0,0 +1,139 @@
+"""Every subprocess must be an argument list, never a shell string."""
+
+import argparse
+import subprocess
+
+import pytest
+
+from dsc import client as client_module
+from dsc import const
+from dsc.client import SeafileClient
+from dsc.misc import hide_password, user_cmd
+
+# Values that would be interpreted by a shell if a command were ever joined
+# into a string and handed to ``su -c``.
+SHELL_METACHARACTERS = [
+    'pa"ss',
+    "pa'ss",
+    "pa$(id)ss",
+    "pa`id`ss",
+    "pass; rm -rf /",
+    "pass && curl evil.example",
+    "pass | tee /tmp/x",
+    "pass\nrm -rf /",
+    "pass with spaces",
+    "pass\\backslash",
+    "паssшörд",
+]
+
+
+@pytest.fixture
+def client():
+    return SeafileClient(
+        host="seafile.example",
+        user="user@example",
+        passwd="s3cret",
+        app_dir="/dsc",
+    )
+
+
+@pytest.fixture
+def calls(monkeypatch):
+    """Capture every subprocess invocation instead of running it."""
+    recorded = []
+
+    def fake_run(argv, **kwargs):
+        recorded.append((argv, kwargs))
+        return subprocess.CompletedProcess(argv, 0, stdout=b"")
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    # Directory creation is covered by the path tests; keep the real
+    # filesystem out of command construction.
+    monkeypatch.setattr(client_module, "create_dir", lambda path: None)
+    return recorded
+
+
+def test_user_cmd_uses_runuser_without_a_shell():
+    argv = user_cmd(["seaf-cli", "status"])
+    assert argv == ["runuser", "-u", const.DEFAULT_USERNAME, "--", "seaf-cli", "status"]
+
+
+def test_user_cmd_never_invokes_a_shell():
+    argv = user_cmd(["seaf-cli", "sync", "-d", "a; rm -rf /"])
+    assert "-c" not in argv
+    assert not any(arg.endswith("sh") for arg in argv)
+    # The dangerous value stays a single, unsplit argument.
+    assert argv[-1] == "a; rm -rf /"
+
+
+def test_user_cmd_rejects_a_string():
+    with pytest.raises(TypeError):
+        user_cmd("seaf-cli status")
+
+
+@pytest.mark.parametrize("password", SHELL_METACHARACTERS)
+def test_sync_passes_password_as_one_unquoted_argument(client, calls, password):
+    client.password = password
+    client.sync_lib("lib-id", "/dsc/seafile/Docs")
+
+    argv, _ = calls[-1]
+    assert argv[argv.index("-p") + 1] == password
+
+
+@pytest.mark.parametrize("lib_dir", [
+    "/dsc/seafile/a\"; touch /tmp/pwned; #",
+    "/dsc/seafile/$(id)",
+    "/dsc/seafile/`id`",
+    "/dsc/seafile/name with spaces",
+])
+def test_sync_passes_library_dir_as_one_argument(client, calls, lib_dir):
+    client.sync_lib("lib-id", lib_dir)
+
+    argv, _ = calls[-1]
+    assert argv[argv.index("-d") + 1] == lib_dir
+
+
+def test_sync_command_is_a_list_of_strings(client, calls):
+    client.sync_lib("lib-id", "/dsc/seafile/Docs")
+
+    argv, kwargs = calls[-1]
+    assert isinstance(argv, list)
+    assert all(isinstance(arg, str) for arg in argv)
+    assert kwargs.get("shell") is not True
+
+
+@pytest.mark.parametrize("password", SHELL_METACHARACTERS)
+def test_sync_never_logs_the_password(client, calls, caplog, password):
+    client.password = password
+    with caplog.at_level("INFO"):
+        client.sync_lib("lib-id", "/dsc/seafile/Docs")
+
+    assert password not in caplog.text
+
+
+def test_hide_password_masks_every_occurrence():
+    masked = hide_password(["-p", "secret", "-u", "secret-user"], "secret")
+    assert masked == ["-p", "********", "-u", "********-user"]
+
+
+def test_hide_password_does_not_mutate_the_original():
+    argv = ["-p", "secret"]
+    hide_password(argv, "secret")
+    assert argv == ["-p", "secret"]
+
+
+def test_configure_passes_option_values_as_arguments(client, calls, monkeypatch):
+    # Changing an option restarts the daemon, and that restart polls without a
+    # timeout. Keep this test on command construction only.
+    monkeypatch.setattr(client, "stop_daemon", lambda: None)
+    monkeypatch.setattr(client, "start_daemon", lambda: None)
+
+    args = argparse.Namespace(upload_limit=100, unrelated="ignored")
+    client.configure(args, check_for_daemon=False)
+
+    argvs = [argv for argv, _ in calls]
+    prefix = ["runuser", "-u", const.DEFAULT_USERNAME, "--", "seaf-cli", "config"]
+    assert prefix + ["-k", "upload_limit"] in argvs
+    assert prefix + ["-k", "upload_limit", "-v", "100"] in argvs
+    # An option that seaf-cli does not know about is never forwarded.
+    assert not any("ignored" in argv for argv in argvs)

tests/test_paths.py

@@ -0,0 +1,111 @@
+import os
+
+import pytest
+
+from dsc.paths import UnsafeLibraryName, lib_dirname, plan_lib_dirs, resolve_lib_dir
+
+PARENT = "/dsc/seafile"
+
+
+@pytest.mark.parametrize("name, expected", [
+    ("Documents", "Documents"),
+    ("My Library", "My_Library"),
+    ("  padded  ", "padded"),
+    ("Ünïcödé ☂", "Ünïcödé_☂"),
+    # The characters below used to reach a shell. They are now ordinary
+    # filename characters and must survive untouched.
+    ('quote"name', 'quote"name'),
+    ("single'name", "single'name"),
+    ("semi;colon", "semi;colon"),
+    ("dollar$(id)", "dollar$(id)"),
+    ("back`id`tick", "back`id`tick"),
+    ("pipe|amp&name", "pipe|amp&name"),
+    ("new\nline", "new_line"),
+    ("tab\tname", "tab_name"),
+    ("dash-lead", "dash-lead"),
+    ("-leading-dash", "-leading-dash"),
+])
+def test_lib_dirname_accepts(name, expected):
+    assert lib_dirname(name) == expected
+
+
+@pytest.mark.parametrize("name", [
+    "",
+    "   ",
+    "\n",
+    ".",
+    "..",
+    "...",  # not special to the kernel, but never a library name we want
+    "/",
+    "/etc",
+    "/etc/passwd",
+    "../escape",
+    "../../root",
+    "nested/name",
+    "trailing/",
+    "nul\0byte",
+])
+def test_lib_dirname_rejects(name):
+    with pytest.raises(UnsafeLibraryName):
+        lib_dirname(name)
+
+
+def test_resolve_lib_dir_stays_under_parent():
+    assert resolve_lib_dir(PARENT, "Documents") == "/dsc/seafile/Documents"
+
+
+def test_resolve_lib_dir_rejects_escape_through_symlink(tmp_path):
+    parent = tmp_path / "seafile"
+    parent.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (parent / "escape").symlink_to(outside)
+
+    with pytest.raises(UnsafeLibraryName):
+        resolve_lib_dir(str(parent), "escape")
+
+
+def test_resolve_lib_dir_allows_real_dir_under_parent(tmp_path):
+    parent = tmp_path / "seafile"
+    parent.mkdir()
+    (parent / "Documents").mkdir()
+
+    resolved = resolve_lib_dir(str(parent), "Documents")
+    assert resolved == os.path.realpath(str(parent / "Documents"))
+
+
+def test_plan_lib_dirs_maps_each_library_to_its_own_dir():
+    plan = plan_lib_dirs({"id-a": "Docs", "id-b": "My Photos"}, PARENT)
+    assert plan == {
+        "id-a": "/dsc/seafile/Docs",
+        "id-b": "/dsc/seafile/My_Photos",
+    }
+
+
+def test_plan_lib_dirs_disambiguates_duplicate_names():
+    libs = {"aaaaaaaabbbb": "Docs", "ccccccccdddd": "Docs"}
+    plan = plan_lib_dirs(libs, PARENT)
+
+    assert plan == {
+        "aaaaaaaabbbb": "/dsc/seafile/Docs-aaaaaaaa",
+        "ccccccccdddd": "/dsc/seafile/Docs-cccccccc",
+    }
+
+
+def test_plan_lib_dirs_duplicate_suffix_is_order_independent():
+    libs = {"aaaaaaaabbbb": "Docs", "ccccccccdddd": "Docs"}
+    reversed_libs = dict(reversed(list(libs.items())))
+    assert plan_lib_dirs(libs, PARENT) == plan_lib_dirs(reversed_libs, PARENT)
+
+
+def test_plan_lib_dirs_duplicates_collide_after_space_folding():
+    """'A B' and 'A_B' both fold to 'A_B' and must not share a directory."""
+    libs = {"aaaaaaaabbbb": "A B", "ccccccccdddd": "A_B"}
+    plan = plan_lib_dirs(libs, PARENT)
+    assert len(set(plan.values())) == 2
+
+
+def test_plan_lib_dirs_skips_unsafe_names_and_keeps_the_rest():
+    libs = {"good": "Docs", "bad": "../escape"}
+    plan = plan_lib_dirs(libs, PARENT)
+    assert plan == {"good": "/dsc/seafile/Docs"}