mirror of
https://github.com/Quad4-Software/MeshChatX.git
synced 2026-08-18 09:49:09 -04:00
538 lines
18 KiB
Python
538 lines
18 KiB
Python
# SPDX-License-Identifier: 0BSD
|
||
|
||
"""Adversarial regression and Hypothesis fuzz coverage for RNS FileSync."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from hypothesis import HealthCheck, given, settings
|
||
from hypothesis import strategies as st
|
||
|
||
from meshchatx.src.backend.rns_filesync_handler import (
|
||
RnsFilesyncHandler,
|
||
_is_forbidden_entry_name,
|
||
)
|
||
from rns_filesync.paths import PathJailError, normalize_relpath
|
||
|
||
_TRAVERSAL_PAYLOADS = (
|
||
"../etc/passwd",
|
||
"..\\windows\\system32",
|
||
"a/../../b",
|
||
"/etc/passwd",
|
||
"C:\\Windows\\System32",
|
||
"a\x00b",
|
||
"",
|
||
".",
|
||
"..",
|
||
"//evil",
|
||
"\\\\evil",
|
||
"foo/../../../etc/shadow",
|
||
".rns-filesync.db",
|
||
"dir/.rns-xfer-temp",
|
||
)
|
||
|
||
_BAD_HASHES = (
|
||
"",
|
||
" ",
|
||
"zz",
|
||
"not-hex",
|
||
"../aabb",
|
||
"a" * 15,
|
||
"g" * 32,
|
||
"\x00" * 16,
|
||
"' OR '1'='1",
|
||
"../../identity",
|
||
"а" * 32,
|
||
"all",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def handler(tmp_path):
|
||
storage = tmp_path / "identity_a"
|
||
storage.mkdir()
|
||
identity = SimpleNamespace(hash=b"\xaa" * 16)
|
||
return RnsFilesyncHandler(
|
||
reticulum_instance=MagicMock(name="reticulum"),
|
||
identity=identity,
|
||
storage_dir=str(storage),
|
||
)
|
||
|
||
|
||
def test_path_traversal_payloads_rejected_by_normalize():
|
||
for payload in _TRAVERSAL_PAYLOADS:
|
||
with pytest.raises(PathJailError):
|
||
normalize_relpath(payload)
|
||
|
||
|
||
def test_download_rejects_path_traversal_before_service(handler):
|
||
"""Handler must jail paths itself, not wait for peer connectivity."""
|
||
service = MagicMock()
|
||
handler.service = service
|
||
for payload in _TRAVERSAL_PAYLOADS:
|
||
if not str(payload).strip():
|
||
result = handler.download_file("bb" * 16, payload)
|
||
assert result["ok"] is False
|
||
assert result["error"] == "path is required"
|
||
service.download_file.assert_not_called()
|
||
continue
|
||
service.reset_mock()
|
||
result = handler.download_file("bb" * 16, payload)
|
||
assert result["ok"] is False, payload
|
||
assert "error" in result
|
||
service.download_file.assert_not_called()
|
||
|
||
|
||
def test_download_and_browse_require_running(handler):
|
||
assert handler.download_file("bb" * 16, "a.txt")["ok"] is False
|
||
assert handler.browse_peer("bb" * 16)["ok"] is False
|
||
assert handler.connect_peer("bb" * 16)["ok"] is False
|
||
|
||
|
||
def test_list_directories_rejects_traversal(handler):
|
||
for payload in ("../etc", "/etc/passwd", handler.storage_dir + "/../"):
|
||
result = handler.list_directories(payload)
|
||
# parent of storage may resolve outside and fail jail
|
||
if result.get("ok"):
|
||
assert result["current"].startswith(handler.storage_dir)
|
||
else:
|
||
assert (
|
||
"identity storage" in result["error"]
|
||
or "not a directory" in result["error"]
|
||
)
|
||
|
||
|
||
def test_create_directory_rejects_dotfiles_and_separators(handler):
|
||
assert handler.create_directory(handler._root, ".hidden")["ok"] is False
|
||
assert handler.create_directory(handler._root, "a/b")["ok"] is False
|
||
assert handler.create_directory("/etc", "nope")["ok"] is False
|
||
|
||
|
||
@pytest.mark.parametrize("bad_hash", _BAD_HASHES)
|
||
def test_connect_rejects_bad_hashes(handler, bad_hash):
|
||
handler.service = MagicMock()
|
||
result = handler.connect_peer(bad_hash)
|
||
assert result["ok"] is False
|
||
assert "invalid" in result["error"] or "required" in result["error"]
|
||
handler.service.connect_peer.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("bad_hash", [h for h in _BAD_HASHES if h != "all"])
|
||
def test_acl_grant_rejects_bad_hashes(handler, bad_hash):
|
||
result = handler.update_acl(
|
||
identity_hash=bad_hash,
|
||
perms=["read"],
|
||
enforce=True,
|
||
)
|
||
assert result["ok"] is False
|
||
assert "invalid" in result["error"]
|
||
acl = handler.get_acl()
|
||
assert bad_hash not in acl["rules"].get("read", [])
|
||
assert "../escape" not in acl["rules"].get("read", [])
|
||
|
||
|
||
def test_acl_grant_accepts_all_alias(handler):
|
||
result = handler.update_acl(identity_hash="all", perms=["read"], enforce=True)
|
||
assert result["ok"] is True
|
||
assert "all" in result["rules"]["read"]
|
||
assert handler.get_acl()["rules"]["read"] == ["all"]
|
||
|
||
|
||
def test_acl_enforce_false_persists_across_reload(tmp_path):
|
||
storage = tmp_path / "id"
|
||
storage.mkdir()
|
||
identity = SimpleNamespace(hash=b"\xaa" * 16)
|
||
peer = "cc" * 16
|
||
first = RnsFilesyncHandler(MagicMock(), identity, str(storage))
|
||
first.update_acl(identity_hash=peer, perms=["read"], enforce=True)
|
||
disabled = first.update_acl(enforce=False)
|
||
assert disabled["ok"] is True
|
||
assert disabled["enforce"] is False
|
||
assert first.get_acl()["enforce"] is False
|
||
assert peer in first.get_acl()["rules"]["read"]
|
||
|
||
second = RnsFilesyncHandler(MagicMock(), identity, str(storage))
|
||
acl = second.get_acl()
|
||
assert acl["enforce"] is False
|
||
assert peer in acl["rules"]["read"]
|
||
|
||
|
||
def test_acl_get_matches_update_result(handler):
|
||
peer = "dd" * 16
|
||
updated = handler.update_acl(
|
||
identity_hash=peer, perms=["read", "write"], enforce=True
|
||
)
|
||
fetched = handler.get_acl()
|
||
assert updated["enforce"] is True
|
||
assert fetched["enforce"] is True
|
||
assert fetched["rules"] == updated["rules"]
|
||
assert peer in fetched["rules"]["read"]
|
||
assert peer in fetched["rules"]["write"]
|
||
|
||
|
||
def test_sync_directory_cannot_escape_identity_storage(handler, tmp_path):
|
||
outside = tmp_path / "other_identity" / "filesync" / "sync"
|
||
outside.mkdir(parents=True)
|
||
result = handler.update_settings(sync_directory=str(outside))
|
||
assert result["ok"] is False
|
||
assert "identity storage" in result["error"]
|
||
|
||
nested = handler.storage_dir + "/filesync/custom"
|
||
ok = handler.update_settings(sync_directory=nested)
|
||
assert ok["ok"] is True
|
||
assert ok["sync_directory"] == nested or ok["sync_directory"].endswith(
|
||
"/filesync/custom",
|
||
)
|
||
|
||
# Identity root and reserved top-level trees must never be sync roots.
|
||
root_reject = handler.update_settings(sync_directory=handler.storage_dir)
|
||
assert root_reject["ok"] is False
|
||
bots = os.path.join(handler.storage_dir, "bots")
|
||
os.makedirs(bots, exist_ok=True)
|
||
bots_reject = handler.update_settings(sync_directory=bots)
|
||
assert bots_reject["ok"] is False
|
||
|
||
|
||
def test_start_rejects_escaped_sync_directory(handler, tmp_path):
|
||
outside = tmp_path / "escape_me"
|
||
outside.mkdir()
|
||
with patch("meshchatx.src.backend.rns_filesync_handler.FileSyncService") as mocked:
|
||
result = handler.start(sync_directory=str(outside))
|
||
assert result["ok"] is False
|
||
mocked.assert_not_called()
|
||
|
||
|
||
def test_teardown_clears_service_and_isolates_storage(tmp_path):
|
||
storage_a = tmp_path / "a"
|
||
storage_b = tmp_path / "b"
|
||
storage_a.mkdir()
|
||
storage_b.mkdir()
|
||
identity_a = SimpleNamespace(hash=b"\xaa" * 16)
|
||
identity_b = SimpleNamespace(hash=b"\xbb" * 16)
|
||
ha = RnsFilesyncHandler(MagicMock(), identity_a, str(storage_a))
|
||
hb = RnsFilesyncHandler(MagicMock(), identity_b, str(storage_b))
|
||
peer = "ee" * 16
|
||
ha.update_acl(identity_hash=peer, perms=["read"], enforce=True)
|
||
hb.update_acl(identity_hash=peer, perms=["write"], enforce=True)
|
||
|
||
svc = MagicMock()
|
||
svc.get_status.return_value = {"running": True}
|
||
ha.service = svc
|
||
ha.teardown()
|
||
assert ha.service is None
|
||
svc.stop.assert_called()
|
||
|
||
assert peer in ha.get_acl()["rules"]["read"]
|
||
assert peer in hb.get_acl()["rules"]["write"]
|
||
assert peer not in ha.get_acl()["rules"].get("write", [])
|
||
assert (storage_a / "filesync" / "acl.txt").is_file()
|
||
assert (storage_b / "filesync" / "acl.txt").is_file()
|
||
|
||
|
||
def test_concurrent_acl_mutations_stable(handler):
|
||
peer = "ff" * 16
|
||
errors: list[BaseException] = []
|
||
|
||
def worker(idx: int):
|
||
try:
|
||
for i in range(40):
|
||
handler.update_acl(
|
||
identity_hash=peer,
|
||
perms=["read"] if (idx + i) % 2 == 0 else ["write"],
|
||
enforce=True,
|
||
)
|
||
acl = handler.get_acl()
|
||
assert acl["enforce"] is True
|
||
assert peer in acl["rules"]["read"] or peer in acl["rules"]["write"]
|
||
except BaseException as exc:
|
||
errors.append(exc)
|
||
|
||
threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(timeout=10)
|
||
assert not errors
|
||
|
||
|
||
@settings(
|
||
max_examples=50,
|
||
deadline=None,
|
||
suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
|
||
)
|
||
@given(path=st.sampled_from(list(_TRAVERSAL_PAYLOADS) + ["ok.txt", "dir/file.bin"]))
|
||
def test_download_path_oracle(handler, path):
|
||
service = MagicMock()
|
||
service.download_file.return_value = {"ok": True, "path": path}
|
||
handler.service = service
|
||
result = handler.download_file("aa" * 16, path)
|
||
assert isinstance(result, dict)
|
||
assert "ok" in result
|
||
if not str(path).strip():
|
||
assert result["ok"] is False
|
||
service.download_file.assert_not_called()
|
||
return
|
||
try:
|
||
normalize_relpath(path)
|
||
except PathJailError:
|
||
assert result["ok"] is False
|
||
service.download_file.assert_not_called()
|
||
else:
|
||
assert result["ok"] is True
|
||
service.download_file.assert_called_once()
|
||
|
||
|
||
@settings(
|
||
max_examples=40,
|
||
deadline=None,
|
||
suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
|
||
)
|
||
@given(
|
||
identity_hash=st.one_of(
|
||
st.sampled_from(list(_BAD_HASHES)),
|
||
st.from_regex(r"[0-9a-f]{32}", fullmatch=True),
|
||
st.text(min_size=0, max_size=40),
|
||
),
|
||
perms=st.lists(
|
||
st.sampled_from(["read", "write", "delete", "admin", "nope", ""]),
|
||
max_size=6,
|
||
),
|
||
)
|
||
def test_acl_hash_oracle(handler, identity_hash, perms):
|
||
effective_perms = perms if perms else ["read"]
|
||
result = handler.update_acl(
|
||
identity_hash=identity_hash,
|
||
perms=effective_perms,
|
||
enforce=True,
|
||
)
|
||
assert isinstance(result, dict)
|
||
assert "ok" in result
|
||
cleaned = str(identity_hash or "").strip().lower().replace(":", "")
|
||
valid = cleaned == "all" or (
|
||
len(cleaned) == 32 and all(c in "0123456789abcdef" for c in cleaned)
|
||
)
|
||
if not valid:
|
||
assert result["ok"] is False
|
||
return
|
||
if not any(p in ("read", "write", "delete") for p in effective_perms):
|
||
assert result["ok"] is False
|
||
return
|
||
assert result["ok"] is True
|
||
assert result["rules"] == handler.get_acl()["rules"]
|
||
|
||
|
||
@settings(
|
||
max_examples=30,
|
||
deadline=None,
|
||
suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
|
||
)
|
||
@given(
|
||
sync_directory=st.sampled_from(
|
||
[
|
||
"",
|
||
" ",
|
||
"../escape",
|
||
"/tmp/x",
|
||
"~/.ssh",
|
||
"filesync/nested",
|
||
"filesync/../filesync/ok",
|
||
],
|
||
),
|
||
)
|
||
def test_settings_path_oracle(handler, sync_directory):
|
||
before = handler.get_status()["sync_directory"]
|
||
result = handler.update_settings(sync_directory=sync_directory)
|
||
assert isinstance(result, dict)
|
||
if result["ok"]:
|
||
assert result["sync_directory"].startswith(handler.storage_dir)
|
||
else:
|
||
assert handler.get_status()["sync_directory"] == before
|
||
|
||
|
||
@patch("meshchatx.src.backend.rns_filesync_handler.FileSyncService")
|
||
def test_start_with_malicious_interval_rejected(mock_service_cls, handler):
|
||
result = handler.start(announce_interval=1)
|
||
assert result["ok"] is False
|
||
mock_service_cls.assert_not_called()
|
||
|
||
|
||
@patch("meshchatx.src.backend.rns_filesync_handler.FileSyncService")
|
||
def test_start_wires_callbacks_and_reuses_host_reticulum(mock_service_cls, handler):
|
||
service = MagicMock()
|
||
service.start.return_value = "ab" * 16
|
||
service.get_status.return_value = {
|
||
"running": True,
|
||
"sync_directory": handler._sync_directory,
|
||
"identity_hash": "aa" * 16,
|
||
"destination_hash": "ab" * 16,
|
||
"peers": 0,
|
||
"files": 0,
|
||
"whitelist": False,
|
||
"monitor": True,
|
||
}
|
||
mock_service_cls.return_value = service
|
||
result = handler.start()
|
||
assert result["ok"] is True
|
||
assert mock_service_cls.call_args.kwargs["own_reticulum"] is False
|
||
assert service.on_error is not None
|
||
assert service.on_sync_progress is not None
|
||
|
||
|
||
def test_manager_rejects_traversal_payloads(handler, tmp_path):
|
||
bait = tmp_path / "bait.txt"
|
||
bait.write_text("do-not-touch", encoding="utf-8")
|
||
identity_dir = os.path.join(handler.storage_dir, "identity")
|
||
os.makedirs(identity_dir, exist_ok=True)
|
||
secret = os.path.join(identity_dir, "secret.key")
|
||
with open(secret, "w", encoding="utf-8") as handle:
|
||
handle.write("private")
|
||
|
||
payloads = list(_TRAVERSAL_PAYLOADS) + [
|
||
str(bait),
|
||
secret,
|
||
os.path.join(handler.storage_dir, "identity"),
|
||
os.path.join(handler.storage_dir, "database.db"),
|
||
os.path.join(handler.storage_dir, "lxmf"),
|
||
handler.storage_dir,
|
||
]
|
||
for payload in payloads:
|
||
tree = handler.list_tree(payload if str(payload).strip() else None)
|
||
if not str(payload).strip():
|
||
assert tree["ok"] is True
|
||
continue
|
||
assert tree["ok"] is False, payload
|
||
|
||
assert handler.manager_content(str(payload))["ok"] is False
|
||
assert handler.manager_delete(str(payload))["ok"] is False
|
||
assert handler.manager_mkdir(str(payload))["ok"] is False
|
||
assert (
|
||
handler.manager_upload(
|
||
filename="x.txt",
|
||
data=b"x",
|
||
subdir=str(payload),
|
||
)["ok"]
|
||
is False
|
||
)
|
||
|
||
assert bait.read_text(encoding="utf-8") == "do-not-touch"
|
||
with open(secret, encoding="utf-8") as handle:
|
||
assert handle.read() == "private"
|
||
|
||
|
||
def test_manager_rejects_cross_identity_paths(tmp_path):
|
||
storage_a = tmp_path / "id_a"
|
||
storage_b = tmp_path / "id_b"
|
||
storage_a.mkdir()
|
||
storage_b.mkdir()
|
||
ha = RnsFilesyncHandler(
|
||
MagicMock(), SimpleNamespace(hash=b"\xaa" * 16), str(storage_a)
|
||
)
|
||
hb = RnsFilesyncHandler(
|
||
MagicMock(), SimpleNamespace(hash=b"\xbb" * 16), str(storage_b)
|
||
)
|
||
|
||
bait = os.path.join(hb._sync_directory, "peer_secret.txt")
|
||
with open(bait, "w", encoding="utf-8") as handle:
|
||
handle.write("b-only")
|
||
|
||
assert ha.list_tree(bait)["ok"] is False
|
||
assert ha.manager_content(bait)["ok"] is False
|
||
assert ha.manager_delete(bait)["ok"] is False
|
||
assert os.path.isfile(bait)
|
||
with open(bait, encoding="utf-8") as handle:
|
||
assert handle.read() == "b-only"
|
||
|
||
|
||
@pytest.mark.skipif(os.name == "nt", reason="symlink tests require POSIX")
|
||
def test_manager_rejects_symlink_escape(handler, tmp_path):
|
||
outside = tmp_path / "outside_secret.txt"
|
||
outside.write_text("escape-me", encoding="utf-8")
|
||
link_path = os.path.join(handler._sync_directory, "escape.txt")
|
||
os.symlink(str(outside), link_path)
|
||
|
||
assert handler.list_tree()["ok"] is True
|
||
names = {e["name"] for e in handler.list_tree()["entries"]}
|
||
assert "escape.txt" not in names
|
||
|
||
assert handler.manager_content("escape.txt")["ok"] is False
|
||
assert handler.manager_delete("escape.txt")["ok"] is False
|
||
assert (
|
||
handler.manager_upload(
|
||
filename="escape.txt",
|
||
data=b"overwrite",
|
||
subdir="",
|
||
)["ok"]
|
||
is False
|
||
)
|
||
assert outside.read_text(encoding="utf-8") == "escape-me"
|
||
|
||
|
||
def test_manager_upload_rejects_malicious_filenames(handler):
|
||
for name in (
|
||
".hidden",
|
||
".rns-filesync.db",
|
||
"",
|
||
".",
|
||
"..",
|
||
"x\x00y.txt",
|
||
):
|
||
result = handler.manager_upload(filename=name, data=b"x")
|
||
assert result["ok"] is False, name
|
||
|
||
# Path segments in the client filename are stripped to a basename under sync root.
|
||
escaped = handler.manager_upload(filename="../evil.txt", data=b"safe")
|
||
assert escaped["ok"] is True
|
||
assert escaped["path"] == "evil.txt"
|
||
assert os.path.isfile(os.path.join(handler._sync_directory, "evil.txt"))
|
||
assert not os.path.exists(os.path.join(handler.storage_dir, "evil.txt"))
|
||
|
||
|
||
def test_manager_refuses_delete_sync_root(handler):
|
||
assert handler.manager_delete("")["ok"] is False
|
||
assert handler.manager_delete(".")["ok"] is False
|
||
assert os.path.isdir(handler._sync_directory)
|
||
|
||
|
||
@settings(
|
||
max_examples=60,
|
||
deadline=None,
|
||
suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
|
||
)
|
||
@given(
|
||
path=st.one_of(
|
||
st.sampled_from(
|
||
list(_TRAVERSAL_PAYLOADS) + ["ok.txt", "dir/file.bin", "nested/a/b"]
|
||
),
|
||
st.text(min_size=0, max_size=40),
|
||
),
|
||
)
|
||
def test_manager_path_oracle(handler, path):
|
||
"""Oracle: manager resolve accepts only normalize_relpath-safe relative paths."""
|
||
cleaned = str(path or "").strip()
|
||
expect_ok = False
|
||
if cleaned and not os.path.isabs(cleaned) and not cleaned.startswith(("/", "\\")):
|
||
try:
|
||
safe = normalize_relpath(cleaned)
|
||
parts = safe.replace("\\", "/").split("/")
|
||
if not any(_is_forbidden_entry_name(part) for part in parts):
|
||
expect_ok = True
|
||
except PathJailError:
|
||
expect_ok = False
|
||
|
||
abspath, err = handler._resolve_manager_path(path, allow_root=False)
|
||
if expect_ok:
|
||
# Path may not exist yet. resolve without must_exist should succeed.
|
||
assert err is None, path
|
||
assert abspath is not None
|
||
assert abspath.startswith(handler._sync_root() + os.sep)
|
||
else:
|
||
if cleaned == "":
|
||
assert err == "path is required"
|
||
else:
|
||
assert abspath is None
|
||
assert err is not None
|