diff --git a/docs/user-guide.md b/docs/user-guide.md index 3d50b2d..c18e80a 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -188,7 +188,9 @@ Also create an `MLFlashpointCheckpointLoader` with those dependencies, and use t # Initialize dependencies (shared singletons) checkpoint_object_manager = CheckpointObjectManager() replication_manager = ReplicationManager() -replication_manager.initialize(checkpoint_object_manager) +# base_dir confines replication to the ML Flashpoint base path: peers can only +# read and write checkpoint objects inside it. +replication_manager.initialize(checkpoint_object_manager, base_dir=mlflashpoint_base_path) checkpoint_loader = DefaultMLFlashpointCheckpointLoader( checkpoint_object_manager=checkpoint_object_manager, @@ -262,7 +264,9 @@ from ml_flashpoint.replication.replication_manager import ReplicationManager # Initialize dependencies (shared singletons) checkpoint_object_manager = CheckpointObjectManager() replication_manager = ReplicationManager() -replication_manager.initialize(checkpoint_object_manager) +# base_dir confines replication to the ML Flashpoint base path: peers can only +# read and write checkpoint objects inside it. +replication_manager.initialize(checkpoint_object_manager, base_dir=mlflashpoint_base_path) # Instantiate the StorageWriter memory_storage_writer = MemoryStorageWriter( diff --git a/src/ml_flashpoint/adapter/nemo/wrapper_util.py b/src/ml_flashpoint/adapter/nemo/wrapper_util.py index 564c70a..70b0a3b 100644 --- a/src/ml_flashpoint/adapter/nemo/wrapper_util.py +++ b/src/ml_flashpoint/adapter/nemo/wrapper_util.py @@ -98,7 +98,7 @@ def wrap_trainer_and_auto_resume_with_mlflashpoint( ckpt_obj_manager = CheckpointObjectManager(pool_config=pool_config) replication_manager = ReplicationManager() - replication_manager.initialize(checkpoint_object_manager=ckpt_obj_manager) + replication_manager.initialize(checkpoint_object_manager=ckpt_obj_manager, base_dir=str(flashpoint_base_container)) ckpt_loader = NeMoMLFlashpointCheckpointLoader( checkpoint_object_manager=ckpt_obj_manager, diff --git a/src/ml_flashpoint/replication/replication_manager.py b/src/ml_flashpoint/replication/replication_manager.py index 48e6575..9190b90 100644 --- a/src/ml_flashpoint/replication/replication_manager.py +++ b/src/ml_flashpoint/replication/replication_manager.py @@ -15,11 +15,12 @@ import abc import concurrent.futures import logging +import os import socket import threading import time from pathlib import Path -from typing import Callable, List, Optional +from typing import Callable, List, Optional, Union import torch import torch.distributed as dist @@ -192,6 +193,7 @@ def initialize( retry_config: ReplicationRetryConfig = get_default_retry_config(), replication_transfer_service: Optional[transfer_service_ext.TransferService] = None, repl_strategy: Optional[ReplicationStrategy] = None, + base_dir: Optional[Union[str, os.PathLike]] = None, ) -> None: """Initializes the ReplicationManager and underlying services. @@ -208,7 +210,24 @@ def initialize( created. repl_strategy: An optional replication strategy. If not provided, a default PairwiseReplicationStrategy will be used. + base_dir: The ML Flashpoint base directory. The TransferService + only reads and writes checkpoint objects inside it: a peer that + asks for, or sends, an object id that resolves outside it (after + following symlinks) gets an error. Required unless + `replication_transfer_service` is provided, in which case that + service's own base directory applies and this argument is + ignored. + + Raises: + ValueError: If `base_dir` is missing and no + `replication_transfer_service` is provided. + RuntimeError: If the TransferService fails to initialize. """ + if replication_transfer_service is None and not base_dir: + raise ValueError( + "base_dir is required when no replication_transfer_service is provided: " + "the TransferService only transfers objects inside it." + ) self._checkpoint_object_manager = checkpoint_object_manager self._listen_port = listen_port self._retry_config = retry_config @@ -220,7 +239,8 @@ def initialize( _LOGGER.info("No TransferService provided, initializing a new one...") self._transfer_service = transfer_service_ext.TransferService() bound_listen_port = self._transfer_service.initialize( - listen_port, + os.fspath(base_dir), + listen_port=listen_port, global_rank=dist.get_rank(), ) if bound_listen_port <= 0: diff --git a/src/ml_flashpoint/replication/transfer_service/bindings.cpp b/src/ml_flashpoint/replication/transfer_service/bindings.cpp index 0031d09..59fed50 100644 --- a/src/ml_flashpoint/replication/transfer_service/bindings.cpp +++ b/src/ml_flashpoint/replication/transfer_service/bindings.cpp @@ -35,13 +35,19 @@ PYBIND11_MODULE(transfer_service_ext, m) { .def(py::init<>()) // Synchronous methods remain unchanged (still blocking) - .def("initialize", &TransferService::Initialize, py::arg("listen_port"), - py::arg("threads") = 16, py::arg("conn_pool_per_peer") = 16, - py::arg("global_rank") = -1, - "Initializes and starts the C++ transfer service. `threads` is the " - "size of each of the two worker pools, one for outbound tasks " - "(async_put/async_get) and one for inbound requests from peers, so " - "the service runs 2 * threads worker threads.") + .def( + "initialize", &TransferService::Initialize, py::arg("base_dir"), + py::arg("listen_port") = 0, py::arg("threads") = 16, + py::arg("conn_pool_per_peer") = 16, py::arg("global_rank") = -1, + "Initializes and starts the C++ transfer service. `base_dir` is the " + "directory that confines every object id the service reads for a " + "peer or writes on its behalf: an id that resolves outside it, after " + "following symlinks, is rejected with an error. It must not be " + "empty and is created if it does not exist. `threads` is the size " + "of each of the two worker pools, one for outbound tasks " + "(async_put/async_get) and one for inbound requests from peers, so " + "the service runs 2 * threads worker threads. Returns the port the " + "service listens on, or -1 on failure.") .def("shutdown", &TransferService::Shutdown, "Stops the C++ transer service.") diff --git a/src/ml_flashpoint/replication/transfer_service/transfer_helpers.cpp b/src/ml_flashpoint/replication/transfer_service/transfer_helpers.cpp index 89f2a40..5583f80 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_helpers.cpp +++ b/src/ml_flashpoint/replication/transfer_service/transfer_helpers.cpp @@ -15,9 +15,17 @@ #include "transfer_helpers.h" #include +#include #include #include #include +#include +#include +#include + +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/str_cat.h" namespace ml_flashpoint::replication::transfer_service { @@ -49,4 +57,37 @@ std::string GenerateUuid() { return ss.str(); } +absl::StatusOr ResolvePathWithinBaseDir( + std::string_view object_id, const std::filesystem::path& base_dir) { + if (object_id.empty()) { + return absl::InvalidArgumentError("Object id is empty"); + } + std::error_code ec; + // weakly_canonical leaves a relative path relative when none of its + // components exist yet, so make it absolute first. + const std::filesystem::path absolute = + std::filesystem::absolute(std::filesystem::path(object_id), ec); + if (ec) { + return absl::InternalError(absl::StrCat("Cannot resolve object id '", + object_id, "': ", ec.message())); + } + // Follows symlinks in the existing prefix, then normalizes the rest, so a + // destination that does not exist yet is handled too. + const std::filesystem::path resolved = + std::filesystem::weakly_canonical(absolute, ec); + if (ec) { + return absl::InternalError(absl::StrCat("Cannot resolve object id '", + object_id, "': ", ec.message())); + } + // Component-wise comparison: a string prefix check would accept + // "/base2/x" for the base "/base". + const std::filesystem::path relative = resolved.lexically_relative(base_dir); + if (relative.empty() || relative == "." || *relative.begin() == "..") { + return absl::InvalidArgumentError(absl::StrCat( + "Object id '", object_id, "' resolves to '", resolved.native(), + "', outside the base directory '", base_dir.native(), "'")); + } + return resolved.native(); +} + } // namespace ml_flashpoint::replication::transfer_service \ No newline at end of file diff --git a/src/ml_flashpoint/replication/transfer_service/transfer_helpers.h b/src/ml_flashpoint/replication/transfer_service/transfer_helpers.h index a6be324..951d817 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_helpers.h +++ b/src/ml_flashpoint/replication/transfer_service/transfer_helpers.h @@ -17,12 +17,34 @@ #ifndef ML_FLASHPOINT_REPLICATION_TRANSFER_HELPERS_H_ #define ML_FLASHPOINT_REPLICATION_TRANSFER_HELPERS_H_ +#include #include +#include + +#include "absl/status/statusor.h" namespace ml_flashpoint::replication::transfer_service { std::string GenerateUuid(); +// Resolves an object id to the path the service may use for it. +// +// The id is made absolute (a relative id is resolved against the process's +// current working directory, as open(2) would), symlinks in the part of it +// that exists are followed, and the rest is normalized lexically, so "..", +// "." and symlinks cannot lead outside the base directory. `base_dir` itself +// is not a valid object path. +// +// `base_dir` must be absolute and canonical, as returned by +// std::filesystem::canonical, so both sides of the comparison name the real +// location. +// +// Returns the resolved path, or InvalidArgument when the id is empty or +// resolves outside `base_dir`, or another error when the filesystem lookup +// fails (for example, a name that is too long). +absl::StatusOr ResolvePathWithinBaseDir( + std::string_view object_id, const std::filesystem::path& base_dir); + } // namespace ml_flashpoint::replication::transfer_service #endif // ML_FLASHPOINT_REPLICATION_TRANSFER_HELPERS_H_ diff --git a/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp b/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp index 6293de4..7e90853 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp +++ b/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include "absl/log/check.h" #include "absl/log/globals.h" @@ -81,8 +82,9 @@ TransferService::~TransferService() { absl::RemoveLogSink(mlf_log_sink_.get()); } -int TransferService::Initialize(int listen_port, int threads, - int conn_pool_size_per_peer, int global_rank) { +int TransferService::Initialize(const std::string& base_dir, int listen_port, + int threads, int conn_pool_size_per_peer, + int global_rank) { std::lock_guard lock(init_mutex_); if (running_.load()) { LOG(INFO) << "Transfer service is already running."; @@ -99,6 +101,28 @@ int TransferService::Initialize(int listen_port, int threads, mlf_log_sink_ = std::make_unique(global_rank); absl::AddLogSink(mlf_log_sink_.get()); + // Resolve the directory that confines every object id. There is no default + // on purpose: a service that is not told where it may read and write must + // not start. canonical() follows symlinks and needs the directory to exist, + // so object ids are compared against the directory's real location. + if (base_dir.empty()) { + LOG(ERROR) << "base_dir must not be empty"; + return -1; + } + std::error_code ec; + std::filesystem::create_directories(base_dir, ec); + if (!ec) { + base_dir_ = std::filesystem::canonical(base_dir, ec); + } + if (!ec && !std::filesystem::is_directory(base_dir_, ec)) { + ec = std::make_error_code(std::errc::not_a_directory); + } + if (ec) { + LOG(ERROR) << "Failed to resolve base_dir '" << base_dir + << "': " << ec.message(); + return -1; + } + // Get listener fd and bind to listen port absl::StatusOr listener_fd_or = SetupListeningSocket(listen_port); if (!listener_fd_or.ok()) { @@ -154,7 +178,8 @@ int TransferService::Initialize(int listen_port, int threads, LOG(INFO) << "TransferService initialized and listening on port " << listen_port_ << " with threads=" << threads_ << ", conn_pool_per_peer=" << conn_pool_size_per_peer_ - << ", global_rank=" << global_rank_; + << ", global_rank=" << global_rank_ + << ", base_dir=" << base_dir_.native(); return listen_port_; } @@ -399,6 +424,17 @@ std::future TransferService::AsyncGet( pending_tasks_[task_id] = {promise, metric_container}; } + // Check the local destination now, before anything is sent, so that a bad + // id fails fast instead of after a full transfer. HandleDataReceive checks + // it again when the data arrives: that is the one place every write goes + // through. + if (absl::StatusOr dest = + ResolvePathWithinBaseDir(dest_obj_id, base_dir_); + !dest.ok()) { + ReportResult(task_id, false, std::string(dest.status().message())); + return future; + } + task_queue_.enqueue(std::move(task)); return future; @@ -801,6 +837,18 @@ void TransferService::HandleDataReceive(int client_fd, fail("Invalid obj_size received: " + std::to_string(header.obj_size)); return; } + // The destination id comes from the peer for a Put, and from the local + // caller for a Get. Either way it must resolve into base_dir_. From here on + // the resolved path is the one used, so what was checked is what is + // written. The sender is already streaming obj_size bytes, so drain them, + // as the other early failures do, to keep the connection usable. + absl::StatusOr dest_path = + ResolvePathWithinBaseDir(header.dest_obj_id, base_dir_); + if (!dest_path.ok()) { + RecvAndDiscard(client_fd, header.obj_size).IgnoreError(); + fail(std::string(dest_path.status().message())); + return; + } if (is_respond_get_task) { UpdateTaskMetrics( header.task_id, [obj_size = header.obj_size](TaskMetricContainer& ts) { @@ -810,8 +858,7 @@ void TransferService::HandleDataReceive(int client_fd, } }); } - std::string tmp_obj_id = - std::string(header.dest_obj_id) + std::string(kTempFileSuffix); + std::string tmp_obj_id = *dest_path + std::string(kTempFileSuffix); // BufferObject's constructor throws when it cannot create the file. std::optional buffer_obj; // Every failure from here on removes the temporary file. obj_size is the @@ -849,11 +896,11 @@ void TransferService::HandleDataReceive(int client_fd, buffer_obj->close(); // Rename the temporary file to the final destination. - if (rename(tmp_obj_id.c_str(), header.dest_obj_id) != 0) { + if (rename(tmp_obj_id.c_str(), dest_path->c_str()) != 0) { const std::string rename_error = std::strerror(errno); remove_tmp(); - fail("Failed to rename temporary file " + tmp_obj_id + " to " + - header.dest_obj_id + ": " + rename_error); + fail("Failed to rename temporary file " + tmp_obj_id + " to " + *dest_path + + ": " + rename_error); return; } if (is_respond_get_task) { @@ -993,8 +1040,9 @@ void TransferService::ExecuteRespondToGetTask(RespondToGetTask* task) { } metric_container.connection_acquired_time = absl::Now(); - // Open the file as a buffer object. The constructor throws if the object - // cannot be opened (unreadable, a directory, or an empty file). The + // Open the file as a buffer object. The path was resolved and checked + // against base_dir_ by HandleGetObjRequest. The constructor throws if the + // object cannot be opened (unreadable, a directory, or an empty file). The // requester is blocked waiting for our response on this socket, so we must // tell it about the failure. std::optional buffer_obj; @@ -1134,11 +1182,23 @@ void TransferService::HandleGetObjRequest(int client_fd, LOG(INFO) << "Handling get object request for requested_obj_id: " << header.source_obj_id; + // The id comes from the peer. Refuse anything that resolves outside + // base_dir_, and hand the resolved path to the task so that what was + // checked is what gets opened. + absl::StatusOr source_path = + ResolvePathWithinBaseDir(header.source_obj_id, base_dir_); + if (!source_path.ok()) { + LOG(ERROR) << "Rejecting get request for '" << header.source_obj_id + << "': " << source_path.status().message(); + SendErrorResponse(client_fd, header.task_id, header.source_obj_id); + return; + } + // Use the non-throwing overload. A stat error other than "not found" (for // example, a name that is too long) must also send kError to the requester // instead of throwing and dropping the socket. std::error_code ec; - if (!std::filesystem::exists(header.source_obj_id, ec)) { + if (!std::filesystem::exists(*source_path, ec)) { LOG(ERROR) << "Object not found: " << header.source_obj_id << (ec ? ": " + ec.message() : ""); SendErrorResponse(client_fd, header.task_id, header.source_obj_id); @@ -1149,16 +1209,15 @@ void TransferService::HandleGetObjRequest(int client_fd, metric_container->task_id = header.task_id; metric_container->task_type = TaskMetricContainer::TaskType::kRespondToGet; try { - metric_container->data_size = - std::filesystem::file_size(header.source_obj_id); + metric_container->data_size = std::filesystem::file_size(*source_path); } catch (const std::filesystem::filesystem_error& e) { LOG(WARNING) << "Failed to get file size for metrics: " << e.what(); metric_container->data_size = 0; } metric_container->submit_time = absl::Now(); - RespondToGetTask task(header.task_id, header.source_obj_id, - header.dest_obj_id, client_fd, metric_container); + RespondToGetTask task(header.task_id, *source_path, header.dest_obj_id, + client_fd, metric_container); // Not registered in pending_tasks_: the task_id is the peer's and could // name a local task. ExecuteRespondToGetTask logs its own outcome. task.Execute(this); diff --git a/src/ml_flashpoint/replication/transfer_service/transfer_service.h b/src/ml_flashpoint/replication/transfer_service/transfer_service.h index 85ca4fc..f2e7b29 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_service.h +++ b/src/ml_flashpoint/replication/transfer_service/transfer_service.h @@ -17,6 +17,7 @@ #ifndef ML_FLASHPOINT_REPLICATION_TRANSFER_SERVICE_H_ #define ML_FLASHPOINT_REPLICATION_TRANSFER_SERVICE_H_ +#include #include #include #include @@ -66,6 +67,13 @@ class TransferService final { // Initializes the transfer service. // // Args: + // base_dir: The directory that confines every object id this service + // reads (kGetObj from a peer) or writes (kPutObj from a peer, + // and the local destination of AsyncGet). An id that resolves + // outside it, after following symlinks, is rejected with + // kError. Must not be empty. The directory is created if it + // does not exist. Relative object ids are resolved against the + // process's current working directory, not against base_dir. // listen_port: The port to listen on. If 0, an ephemeral port is chosen. // threads: The number of worker threads in *each* of the two thread pools // (outbound tasks and inbound epoll events), i.e. 2 * threads @@ -74,9 +82,11 @@ class TransferService final { // global_rank: The global rank of the process. // // Returns: - // The port the service is listening on, or -1 on failure. - int Initialize(int listen_port = 0, int threads = 16, - int conn_pool_per_peer = 16, int global_rank = -1); + // The port the service is listening on, or -1 on failure, including an + // empty base_dir or one that cannot be created or is not a directory. + int Initialize(const std::string& base_dir, int listen_port = 0, + int threads = 16, int conn_pool_per_peer = 16, + int global_rank = -1); // Shuts down the transfer service gracefully. void Shutdown(); @@ -88,7 +98,8 @@ class TransferService final { // data_size: Size of the data in bytes. // dest_address: The destination service address, expected in "host:port" // format (e.g., "127.0.0.1:12345"). - // dest_obj_id: A unique identifier for the object at the destination. + // dest_obj_id: A unique identifier for the object at the destination. The + // destination checks it against its own base directory. // // Returns: // A `std::future` that will be set when the transfer @@ -100,10 +111,13 @@ class TransferService final { // Asynchronously requests an object from the specified peer. // // Args: - // source_obj_id: The ID of the object to request from the remote peer. + // source_obj_id: The ID of the object to request from the remote peer. The + // peer checks it against its own base directory. // source_address: The source service address, expected in "host:port" // format (e.g., "127.0.0.1:12345"). - // dest_obj_id: The ID to assign to the received object locally. + // dest_obj_id: The ID to assign to the received object locally. It must + // be inside this service's base directory; otherwise the + // returned future fails before anything is sent. // // Returns: // A `std::future` that will be set when the transfer @@ -167,6 +181,9 @@ class TransferService final { local_ip_address_; // User input local ip address, only used for testing. std::string local_address_; // Local address used in data transfer. int global_rank_ = -1; + // Absolute, canonical directory that every object id must resolve into. Set + // by Initialize() before running_ becomes true and read-only afterwards. + std::filesystem::path base_dir_; // Two separate pools are required. An outbound task (Put/Get) blocks a // worker for the whole transfer, including the wait for the peer's diff --git a/tests/replication/test_replication_manager.py b/tests/replication/test_replication_manager.py index eb9e3dc..b63e873 100644 --- a/tests/replication/test_replication_manager.py +++ b/tests/replication/test_replication_manager.py @@ -55,11 +55,11 @@ def mock_gather(): replication_manager._gather_replication_service_addresses = mocker.MagicMock(side_effect=mock_gather) # When - replication_manager.initialize(mock_checkpoint_manager, listen_port=0) + replication_manager.initialize(mock_checkpoint_manager, listen_port=0, base_dir="/flashpoint/base") # Then assert replication_manager._listen_port == 12345 - mock_transfer_service_instance.initialize.assert_called_once_with(0, global_rank=0) + mock_transfer_service_instance.initialize.assert_called_once_with("/flashpoint/base", listen_port=0, global_rank=0) def test_initialize_binds_port_failure(replication_manager, mocker): @@ -78,7 +78,42 @@ def test_initialize_binds_port_failure(replication_manager, mocker): # When/Then with pytest.raises(RuntimeError, match="Failed to initialize TransferService"): + replication_manager.initialize(mock_checkpoint_manager, base_dir="/flashpoint/base") + + +def test_initialize_requires_base_dir_when_creating_transfer_service(replication_manager, mocker): + """Tests that initialize refuses to create a TransferService without a base directory to confine it to.""" + # Given + replication_manager._transfer_service = None + mock_transfer_service_cls = mocker.patch( + "ml_flashpoint.replication.transfer_service.transfer_service_ext.TransferService" + ) + mock_checkpoint_manager = mocker.MagicMock() + + # When/Then + with pytest.raises(ValueError, match="base_dir is required"): replication_manager.initialize(mock_checkpoint_manager) + mock_transfer_service_cls.assert_not_called() + + +def test_initialize_accepts_preconfigured_transfer_service_without_base_dir(replication_manager, mocker): + """Tests that a pre-configured TransferService is used as-is, without requiring a base directory.""" + # Given + expected_transfer_service = mocker.MagicMock() + mock_checkpoint_manager = mocker.MagicMock() + replication_manager._initialize_address = mocker.MagicMock() + repl_strategy = mocker.MagicMock(spec=["get_destination_addresses"]) + + # When + replication_manager.initialize( + mock_checkpoint_manager, + replication_transfer_service=expected_transfer_service, + repl_strategy=repl_strategy, + ) + + # Then + assert replication_manager._transfer_service is expected_transfer_service + expected_transfer_service.initialize.assert_not_called() def test_async_replicate_no_strategy(replication_manager, mocker): diff --git a/tests/replication/test_replication_manager_e2e.py b/tests/replication/test_replication_manager_e2e.py index 002f109..6231417 100644 --- a/tests/replication/test_replication_manager_e2e.py +++ b/tests/replication/test_replication_manager_e2e.py @@ -25,28 +25,34 @@ from ml_flashpoint.replication.transfer_service import transfer_service_ext -def run_service(service, port_container, rank): +def run_service(service, port_container, rank, base_dir): """Target function for running a TransferService in a thread.""" - port = service.initialize(0, global_rank=rank) + port = service.initialize(base_dir, global_rank=rank) port_container.append(port) @pytest.fixture -def services(): - """Fixture to set up and tear down sender and receiver TransferServices.""" +def services(tmp_path): + """Fixture to set up and tear down sender and receiver TransferServices. + + Both services are confined to `tmp_path`, where the tests place their objects. + """ sender_service = transfer_service_ext.TransferService() receiver_service = transfer_service_ext.TransferService() sender_port_container = [] receiver_port_container = [] + base_dir = str(tmp_path) # Start receiver service to get its port - receiver_thread = threading.Thread(target=run_service, args=(receiver_service, receiver_port_container, 1)) + receiver_thread = threading.Thread( + target=run_service, args=(receiver_service, receiver_port_container, 1, base_dir) + ) receiver_thread.daemon = True receiver_thread.start() # Start sender service - sender_thread = threading.Thread(target=run_service, args=(sender_service, sender_port_container, 0)) + sender_thread = threading.Thread(target=run_service, args=(sender_service, sender_port_container, 0, base_dir)) sender_thread.daemon = True sender_thread.start() diff --git a/tests/replication/test_transer_service.py b/tests/replication/test_transer_service.py index b776e62..3dfab1e 100644 --- a/tests/replication/test_transer_service.py +++ b/tests/replication/test_transer_service.py @@ -57,7 +57,9 @@ def safe_remove_file(filepath: str) -> None: @pytest.fixture(name="transfer_services") -def fixture_transfer_services() -> tuple[ +def fixture_transfer_services( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> tuple[ transfer_service_ext.TransferService, transfer_service_ext.TransferService, str, @@ -66,7 +68,10 @@ def fixture_transfer_services() -> tuple[ """Initializes and yields two TransferService instances for testing. This fixture handles the lifecycle of two services, ensuring they are - properly initialized before a test runs and shut down afterward. + properly initialized before a test runs and shut down afterward. Both + services use `tmp_path` as their base directory, and the working directory + is switched to it, so the relative object ids the tests use resolve inside + the base directory and no test files are left in the repository. Yields: A tuple containing: @@ -75,6 +80,7 @@ def fixture_transfer_services() -> tuple[ - The network address (host:port) of the first service. - The network address (host:port) of the second service. """ + monkeypatch.chdir(tmp_path) service1 = transfer_service_ext.TransferService() service2 = transfer_service_ext.TransferService() @@ -83,8 +89,8 @@ def fixture_transfer_services() -> tuple[ addr1 = f"127.0.0.1:{port1}" addr2 = f"127.0.0.1:{port2}" - service1.initialize(listen_port=port1) - service2.initialize(listen_port=port2) + service1.initialize(str(tmp_path), listen_port=port1) + service2.initialize(str(tmp_path), listen_port=port2) yield service1, service2, addr1, addr2 @@ -353,6 +359,82 @@ def test_async_get_non_existent_object( get_future.result(timeout=10) +def test_async_put_outside_base_dir_is_rejected( + transfer_services: tuple[ + transfer_service_ext.TransferService, + transfer_service_ext.TransferService, + str, + str, + ], + tmp_path_factory: pytest.TempPathFactory, +) -> None: + """Verifies that the receiving service rejects a destination outside its base directory and writes nothing.""" + # Given + service1, _, _, addr2 = transfer_services + outside_dir = tmp_path_factory.mktemp("outside_put") + dest_object_id = str(outside_dir / "escaped.object") + data_to_send = np.arange(20, dtype=np.int64) + + # When + put_future = service1.async_put(data_to_send.ctypes.data, data_to_send.nbytes, addr2, dest_object_id) + + # Then + with pytest.raises(RuntimeError, match="Received error from destination"): + put_future.result(timeout=10) + assert not pathlib.Path(dest_object_id).exists() + assert not pathlib.Path(dest_object_id + ".tmp").exists() + + +def test_async_get_of_object_outside_peer_base_dir_is_rejected( + transfer_services: tuple[ + transfer_service_ext.TransferService, + transfer_service_ext.TransferService, + str, + str, + ], + tmp_path_factory: pytest.TempPathFactory, +) -> None: + """Verifies that a peer refuses to serve a readable file that lies outside its base directory.""" + # Given + _, service2, addr1, _ = transfer_services + outside_dir = tmp_path_factory.mktemp("outside_get") + secret_path = outside_dir / "secret.object" + secret_path.write_bytes(b"not for peers") + dest_object_id = "fetched_secret.object" + + # When + get_future = service2.async_get(str(secret_path), addr1, dest_object_id) + + # Then + with pytest.raises(RuntimeError, match="Received error message"): + get_future.result(timeout=10) + assert not pathlib.Path(dest_object_id).exists() + + +def test_async_get_with_local_destination_outside_base_dir_fails_fast( + transfer_services: tuple[ + transfer_service_ext.TransferService, + transfer_service_ext.TransferService, + str, + str, + ], + tmp_path_factory: pytest.TempPathFactory, +) -> None: + """Verifies that async_get fails with the reason when the local destination is outside the base directory.""" + # Given + _, service2, addr1, _ = transfer_services + outside_dir = tmp_path_factory.mktemp("outside_dest") + dest_object_id = str(outside_dir / "escaped.object") + + # When + get_future = service2.async_get("any_source.object", addr1, dest_object_id) + + # Then + with pytest.raises(RuntimeError, match="outside the base directory"): + get_future.result(timeout=10) + assert not pathlib.Path(dest_object_id).exists() + + def test_concurrent_and_repeated_async_get( transfer_services: tuple[ transfer_service_ext.TransferService, diff --git a/tests/replication/transfer_service/transfer_helpers_test.cpp b/tests/replication/transfer_service/transfer_helpers_test.cpp index cff86c1..a7c8843 100644 --- a/tests/replication/transfer_service/transfer_helpers_test.cpp +++ b/tests/replication/transfer_service/transfer_helpers_test.cpp @@ -14,13 +14,225 @@ #include "transfer_helpers.h" +#include #include +#include +#include #include +#include + +#include "absl/status/status.h" namespace ml_flashpoint::replication::transfer_service { namespace { + +using ::testing::HasSubstr; + +// Creates a fresh directory tree for each test: +// /base/ the base directory (canonical path in base_) +// /base/inner.txt a regular file inside the base +// /outside.txt a regular file outside the base +class ResolvePathWithinBaseDirTest : public ::testing::Test { + protected: + void SetUp() override { + root_ = std::filesystem::temp_directory_path() / + ("resolve_path_test_" + GenerateUuid()); + std::filesystem::create_directories(root_ / "base"); + std::ofstream(root_ / "base" / "inner.txt") << "inner"; + std::ofstream(root_ / "outside.txt") << "outside"; + root_ = std::filesystem::canonical(root_); + base_ = root_ / "base"; + } + + void TearDown() override { std::filesystem::remove_all(root_); } + + std::filesystem::path root_; + std::filesystem::path base_; +}; + +TEST_F(ResolvePathWithinBaseDirTest, AcceptsExistingFileInsideBase) { + // Tests that a plain path inside the base resolves to itself. + // Given + const std::string object_id = (base_ / "inner.txt").native(); + + // When + const auto actual = ResolvePathWithinBaseDir(object_id, base_); + + // Then + ASSERT_TRUE(actual.ok()) << actual.status(); + EXPECT_EQ(object_id, *actual); +} + +TEST_F(ResolvePathWithinBaseDirTest, AcceptsPathThatDoesNotExistYet) { + // Tests that a destination in a directory that is not created yet is + // accepted, because the receiving side creates files before they exist. + // Given + const std::string object_id = (base_ / "new_dir" / "new.obj").native(); + + // When + const auto actual = ResolvePathWithinBaseDir(object_id, base_); + + // Then + ASSERT_TRUE(actual.ok()) << actual.status(); + EXPECT_EQ(object_id, *actual); +} + +TEST_F(ResolvePathWithinBaseDirTest, NormalizesDotSegmentsThatStayInside) { + // Tests that "." and ".." segments are collapsed when the result is still + // inside the base. + // Given + const std::string object_id = + (base_ / "new_dir" / ".." / "." / "inner.txt").native(); + + // When + const auto actual = ResolvePathWithinBaseDir(object_id, base_); + + // Then + ASSERT_TRUE(actual.ok()) << actual.status(); + EXPECT_EQ((base_ / "inner.txt").native(), *actual); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsParentTraversal) { + // Tests that ".." leading out of the base is rejected, whether or not the + // escaped target exists. + // Given + const std::string existing = (base_ / ".." / "outside.txt").native(); + const std::string missing = (base_ / "a" / ".." / ".." / "nope").native(); + + // When + const auto actual_existing = ResolvePathWithinBaseDir(existing, base_); + const auto actual_missing = ResolvePathWithinBaseDir(missing, base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, + actual_existing.status().code()); + EXPECT_THAT(std::string(actual_existing.status().message()), + HasSubstr("outside the base directory")); + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual_missing.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsAbsolutePathOutsideBase) { + // Tests that an absolute path elsewhere on the filesystem is rejected. + // When + const auto actual = + ResolvePathWithinBaseDir((root_ / "outside.txt").native(), base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsSiblingWithSameNamePrefix) { + // Tests that the check is component-wise: "2/x" is not inside + // "" even though it starts with the same characters. + // Given + const std::filesystem::path sibling = root_ / "base2"; + std::filesystem::create_directories(sibling); + const std::string object_id = (sibling / "x.obj").native(); + + // When + const auto actual = ResolvePathWithinBaseDir(object_id, base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsBaseDirItselfAndEmptyId) { + // Tests that the base directory is not a valid object path and that an + // empty id is rejected before any filesystem access. + // When + const auto actual_base = ResolvePathWithinBaseDir(base_.native(), base_); + const auto actual_trailing_slash = + ResolvePathWithinBaseDir(base_.native() + "/", base_); + const auto actual_empty = ResolvePathWithinBaseDir("", base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual_base.status().code()); + EXPECT_EQ(absl::StatusCode::kInvalidArgument, + actual_trailing_slash.status().code()); + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual_empty.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsSymlinkInsideBasePointingOutside) { + // Tests that a symlink placed inside the base cannot be used to reach a + // file outside it: the link is followed before the check. + // Given + const std::filesystem::path link = base_ / "escape.txt"; + std::filesystem::create_symlink(root_ / "outside.txt", link); + + // When + const auto actual = ResolvePathWithinBaseDir(link.native(), base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual.status().code()); + EXPECT_THAT(std::string(actual.status().message()), + HasSubstr((root_ / "outside.txt").native())); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsSymlinkedDirectoryPointingOutside) { + // Tests that a path through a symlinked directory is resolved through the + // link, so a directory link to the outside is rejected even when the final + // component does not exist yet. + // Given + const std::filesystem::path link_dir = base_ / "dir_link"; + std::filesystem::create_directory_symlink(root_, link_dir); + + // When + const auto actual = + ResolvePathWithinBaseDir((link_dir / "new.obj").native(), base_); + + // Then + EXPECT_EQ(absl::StatusCode::kInvalidArgument, actual.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, FollowsSymlinkThatStaysInside) { + // Tests that a symlink whose target is inside the base is accepted and + // resolved to the target. + // Given + const std::filesystem::path link = base_ / "alias.txt"; + std::filesystem::create_symlink(base_ / "inner.txt", link); + + // When + const auto actual = ResolvePathWithinBaseDir(link.native(), base_); + + // Then + ASSERT_TRUE(actual.ok()) << actual.status(); + EXPECT_EQ((base_ / "inner.txt").native(), *actual); +} + +TEST_F(ResolvePathWithinBaseDirTest, ResolvesRelativeIdAgainstWorkingDir) { + // Tests that a relative id is resolved against the current working + // directory, so it is accepted when that directory is the base and rejected + // when it is not. + // Given + const std::filesystem::path cwd = std::filesystem::current_path(); + const std::string object_id = "relative_id.obj"; + + // When + const auto actual_cwd_base = ResolvePathWithinBaseDir(object_id, cwd); + const auto actual_other_base = ResolvePathWithinBaseDir(object_id, base_); + + // Then + ASSERT_TRUE(actual_cwd_base.ok()) << actual_cwd_base.status(); + EXPECT_EQ((cwd / object_id).native(), *actual_cwd_base); + EXPECT_EQ(absl::StatusCode::kInvalidArgument, + actual_other_base.status().code()); +} + +TEST_F(ResolvePathWithinBaseDirTest, RejectsOverlongComponentWithError) { + // Tests that a filesystem lookup failure (a component longer than NAME_MAX) + // is reported as an error instead of being accepted. + // Given + const std::string object_id = (base_ / std::string(300, 'a')).native(); + + // When + const auto actual = ResolvePathWithinBaseDir(object_id, base_); + + // Then + EXPECT_FALSE(actual.ok()); +} + TEST(TransferHelpersTest, GenerateUuidIsValid) { std::string uuid = GenerateUuid(); diff --git a/tests/replication/transfer_service/transfer_service_p2p_test.cpp b/tests/replication/transfer_service/transfer_service_p2p_test.cpp index 758a571..1358215 100644 --- a/tests/replication/transfer_service/transfer_service_p2p_test.cpp +++ b/tests/replication/transfer_service/transfer_service_p2p_test.cpp @@ -42,6 +42,7 @@ #include "gtest/gtest.h" #include "net_util.h" #include "protocol.h" +#include "transfer_helpers.h" #include "transfer_service.h" namespace ml_flashpoint::replication::transfer_service { @@ -111,13 +112,17 @@ void VerifyFileContentAndRemove(const std::string& file_path, std::remove(file_path.c_str()); } +// Base directory for the services in these tests: the working directory, in +// which the tests create and clean up their object files by relative name. +std::string CwdBaseDir() { return std::filesystem::current_path().native(); } + TEST(TransferServiceP2PTest, SimplePut) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "Hello, world!"; @@ -137,11 +142,11 @@ TEST(TransferServiceP2PTest, SimplePut) { TEST(TransferServiceP2PTest, PutLargeObject) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); // Create a large string (5MB) @@ -167,11 +172,11 @@ TEST(TransferServiceP2PTest, PutLargeObject) { TEST(TransferServiceP2PTest, ShutdownInterruptsTransfer) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); // Create a large string (100MB) to make sure it takes time to send @@ -209,11 +214,11 @@ TEST(TransferServiceP2PTest, ShutdownInterruptsTransfer) { TEST(TransferServiceP2PTest, AsyncPutLargeMmapData) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); const size_t large_size = 1UL * 1024 * 1024 * 1024; // 1 GB @@ -267,11 +272,11 @@ TEST(TransferServiceP2PTest, AsyncPutLargeMmapData) { TEST(TransferServiceP2PTest, PutEmptyObjectShouldFail) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = ""; @@ -303,11 +308,11 @@ TEST(TransferServiceP2PTest, PutEmptyObjectShouldFail) { TEST(TransferServiceP2PTest, PutZeroSizeObjectShouldFail) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "some data"; @@ -329,11 +334,11 @@ TEST(TransferServiceP2PTest, PutZeroSizeObjectShouldFail) { TEST(TransferServiceP2PTest, ConcurrentPut) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); const int num_threads = 8; @@ -370,11 +375,11 @@ TEST(TransferServiceP2PTest, ConcurrentPut) { TEST(TransferServiceP2PTest, SimpleGet) { TransferService service1(std::optional("127.0.0.1")); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2(std::optional("127.0.0.1")); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "Hello, world!"; @@ -402,11 +407,11 @@ TEST(TransferServiceP2PTest, SimpleGet) { TEST(TransferServiceP2PTest, GetLargeObject) { TransferService service1(std::optional("127.0.0.1")); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2(std::optional("127.0.0.1")); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); const size_t large_size = 5 * 1024 * 1024; @@ -442,11 +447,11 @@ TEST(TransferServiceP2PTest, GetLargeObject) { TEST(TransferServiceP2PTest, ConcurrentGet) { TransferService service1(std::optional("127.0.0.1")); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2(std::optional("127.0.0.1")); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); const int num_threads = 8; @@ -492,11 +497,11 @@ TEST(TransferServiceP2PTest, ConcurrentGet) { TEST(TransferServiceP2PTest, GetNonExistentObjectShouldFail) { TransferService service1(std::optional("127.0.0.1")); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2(std::optional("127.0.0.1")); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string obj_id = "non_existent_object"; @@ -528,11 +533,11 @@ TEST(TransferServiceP2PTest, GetNonExistentObjectShouldFail) { TEST(TransferServiceP2PTest, PutCreatesTemporaryFileAndRenames) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "Temporary file test data."; @@ -562,11 +567,11 @@ TEST(TransferServiceP2PTest, PutCreatesTemporaryFileAndRenames) { TEST(TransferServiceP2PTest, GetCreatesTemporaryFileAndRenames) { TransferService service1(std::optional("127.0.0.1")); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2(std::optional("127.0.0.1")); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "Temporary file test data for GET."; @@ -605,11 +610,11 @@ TEST(TransferServiceP2PTest, GetCreatesTemporaryFileAndRenames) { TEST(TransferServiceP2PTest, PutOverwritesEmptyTempFile) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "some data here"; @@ -631,11 +636,11 @@ TEST(TransferServiceP2PTest, PutOverwritesEmptyTempFile) { TEST(TransferServiceP2PTest, PutOverwritesNonEmptyTempFile) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string old_data = "old data"; @@ -658,11 +663,11 @@ TEST(TransferServiceP2PTest, PutOverwritesNonEmptyTempFile) { TEST(TransferServiceP2PTest, PutReplacesExistingFile) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string obj_id = "existing_object"; @@ -687,11 +692,11 @@ TEST(TransferServiceP2PTest, PutReplacesExistingFile) { TEST(TransferServiceP2PTest, PutFailsInRenameWhenTargetExistAsADirectory) { TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "some data"; @@ -721,11 +726,11 @@ TEST(TransferServiceP2PTest, PutFailsInRenameWhenTargetExistAsADirectory) { TEST(TransferServiceP2PTest, GetOverwritesEmptyTempFile) { TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "some data here"; @@ -752,11 +757,11 @@ TEST(TransferServiceP2PTest, GetOverwritesEmptyTempFile) { TEST(TransferServiceP2PTest, GetOverwritesNonEmptyTempFile) { TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string old_data = "old data"; @@ -784,11 +789,11 @@ TEST(TransferServiceP2PTest, GetOverwritesNonEmptyTempFile) { TEST(TransferServiceP2PTest, GetReplacesExistingFile) { TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string old_data = "old data"; @@ -816,11 +821,11 @@ TEST(TransferServiceP2PTest, GetReplacesExistingFile) { TEST(TransferServiceP2PTest, GetFailsInRenameWhenTargetExistAsADirectory) { TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string data = "some data"; @@ -858,11 +863,11 @@ TEST(TransferServiceP2PTest, TimestampsAreRecorded) { absl::AddLogSink(&sink); TransferService service1; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2; - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); // Perform a Put @@ -1153,7 +1158,7 @@ class FakeGetResponder { TEST(TransferServiceP2PTest, GetObjDoesNotConnectToDestAddress) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); std::string expected_data = "ssrf_test_payload"; @@ -1227,8 +1232,9 @@ class SpoofedDestAddressTest : public ::testing::TestWithParam {}; TEST_P(SpoofedDestAddressTest, GetResponseGoesBackOnRequestSocket) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string expected_data = "spoofed_dest_payload"; const std::string source_obj_id = "spoofed_dest_source_obj"; @@ -1279,8 +1285,9 @@ INSTANTIATE_TEST_SUITE_P( TEST(TransferServiceP2PTest, GetRequestWithUnterminatedAddressFieldsIsServed) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string expected_data = "unterminated_payload"; const std::string source_obj_id = "unterminated_source_obj"; @@ -1314,8 +1321,9 @@ TEST(TransferServiceP2PTest, GetRequestForUnopenableObjectReturnsErrorOnSameSocket) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string empty_obj_id = "unopenable_empty_obj"; const std::string dir_obj_id = "unopenable_dir_obj"; @@ -1362,8 +1370,9 @@ TEST(TransferServiceP2PTest, UnsolicitedRespondToGetObjClosesConnectionWithoutWritingFile) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string dest_obj_id = "unsolicited_dest_obj"; std::remove(dest_obj_id.c_str()); @@ -1396,8 +1405,9 @@ TEST(TransferServiceP2PTest, TEST(TransferServiceP2PTest, UnknownMessageTypeClosesConnection) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); int client_fd = ConnectToLocalPort(port1); ASSERT_GE(client_fd, 0); @@ -1420,8 +1430,9 @@ TEST(TransferServiceP2PTest, UnknownMessageTypeClosesConnection) { TEST(TransferServiceP2PTest, UnsolicitedAckOnListenerClosesConnection) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); int client_fd = ConnectToLocalPort(port1); ASSERT_GE(client_fd, 0); @@ -1452,8 +1463,9 @@ TEST(TransferServiceP2PTest, UnsolicitedErrorOnListenerDoesNotFailLocalTask) { int peer_listener = ListenOnLocalEphemeralPort(&peer_port); ASSERT_GE(peer_listener, 0); TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); std::future actual_future = service1.AsyncGet( "any_source_obj", "127.0.0.1:" + std::to_string(peer_port), dest_obj_id); @@ -1518,8 +1530,9 @@ TEST(TransferServiceP2PTest, GetRequestNamingLocalTaskIdDoesNotTouchLocalTask) { int peer_listener = ListenOnLocalEphemeralPort(&peer_port); ASSERT_GE(peer_listener, 0); TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); std::future actual_future = service1.AsyncGet( "any_source_obj", "127.0.0.1:" + std::to_string(peer_port), dest_obj_id); @@ -1575,8 +1588,9 @@ TEST(TransferServiceP2PTest, GetRequestNamingLocalTaskIdDoesNotTouchLocalTask) { TEST(TransferServiceP2PTest, ShutdownWhileResponderWaitsForAckIsSafe) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string expected_data = "shutdown_no_ack_payload"; const std::string source_obj_id = "shutdown_no_ack_source_obj"; @@ -1605,7 +1619,7 @@ TEST(TransferServiceP2PTest, ShutdownWhileResponderWaitsForAckIsSafe) { close(client_fd); } TransferService service2("127.0.0.1"); - EXPECT_GT(service2.Initialize(), 0); + EXPECT_GT(service2.Initialize(CwdBaseDir()), 0); service2.Shutdown(); std::remove(source_obj_id.c_str()); } @@ -1616,8 +1630,9 @@ TEST(TransferServiceP2PTest, ShutdownWhileResponderWaitsForAckIsSafe) { TEST(TransferServiceP2PTest, ShutdownWhilePeerIsNotReadingGetResponseIsSafe) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); // Large enough to exceed any socket buffer autotuning on loopback. const std::string large_data(16 * 1024 * 1024, 'L'); @@ -1651,10 +1666,10 @@ TEST(TransferServiceP2PTest, ShutdownWhilePeerIsNotReadingGetResponseIsSafe) { TEST(TransferServiceP2PTest, GetSucceedsWhenRequesterAdvertisesUnreachableIp) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("203.0.113.5"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); const std::string expected_data = "reachable after all"; const std::string source_obj_id = "unreachable_requester_source_obj"; @@ -1683,8 +1698,9 @@ TEST(TransferServiceP2PTest, GetSucceedsWhenRequesterAdvertisesUnreachableIp) { TEST(TransferServiceP2PTest, SequentialGetRequestsOnOneSocketAreServedInOrder) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const int kNumRequests = 3; std::vector source_obj_ids; @@ -1731,8 +1747,9 @@ TEST(TransferServiceP2PTest, SequentialGetRequestsOnOneSocketAreServedInOrder) { TEST(TransferServiceP2PTest, ConcurrentGetRequestsBeyondThreadCountAreServed) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const int kNumClients = 8; std::vector source_obj_ids(kNumClients); @@ -1788,8 +1805,9 @@ TEST(TransferServiceP2PTest, ConcurrentGetRequestsBeyondThreadCountAreServed) { TEST(TransferServiceP2PTest, ResponderSurvivesClientDisconnectMidResponse) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/1, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/1, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); // Large enough to exceed any socket buffer autotuning on loopback. const std::string large_data(16 * 1024 * 1024, 'L'); @@ -1837,8 +1855,9 @@ TEST(TransferServiceP2PTest, ResponderSurvivesClientDisconnectMidResponse) { TEST(TransferServiceP2PTest, ResponderSurvivesClientClosingInsteadOfAck) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/1, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/1, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string expected_data = "no_ack_then_next_payload"; const std::string source_obj_id = "no_ack_next_source_obj"; @@ -1884,8 +1903,9 @@ class InvalidGetRequestTest : public ::testing::TestWithParam {}; TEST_P(InvalidGetRequestTest, RepliesWithErrorAndKeepsSocketUsable) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string expected_data = "valid_after_invalid_request"; const std::string valid_obj_id = "invalid_request_valid_source_obj"; @@ -1930,8 +1950,9 @@ INSTANTIATE_TEST_SUITE_P(TransferServiceP2PTest, InvalidGetRequestTest, TEST(TransferServiceP2PTest, ReplyHeadersCarryNoBytesAfterTerminators) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string source_obj_id = "clean_header_source_obj"; WriteTestFile(source_obj_id, "clean header payload"); @@ -1971,8 +1992,9 @@ TEST(TransferServiceP2PTest, ReplyHeadersCarryNoBytesAfterTerminators) { TEST(TransferServiceP2PTest, TruncatedPutLeavesNoTemporaryFile) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string dest_obj_id = "truncated_put_dest_obj"; const std::string tmp_obj_id = dest_obj_id + ".tmp"; @@ -2010,8 +2032,9 @@ TEST(TransferServiceP2PTest, PutToUncreatableDestinationRepliesWithErrorAndKeepsSocketUsable) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string undeliverable_payload(256 * 1024, 'U'); // A regular file as a path component makes the destination impossible to @@ -2065,12 +2088,14 @@ TEST(TransferServiceP2PTest, PutFailsCleanlyAndReusesConnectionWhenDestinationCannotBeCreated) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port2 = + service2.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port2, 0); const std::string peer1_addr = "127.0.0.1:" + std::to_string(port1); std::string undeliverable_payload(256 * 1024, 'U'); @@ -2111,12 +2136,14 @@ TEST(TransferServiceP2PTest, GetFailsCleanlyAndReusesConnectionWhenDestinationCannotBeCreated) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1); + int port2 = + service2.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port2, 0); const std::string peer1_addr = "127.0.0.1:" + std::to_string(port1); const std::string large_obj_id = "api_get_uncreatable_large_source_obj"; @@ -2185,8 +2212,9 @@ TEST_P(MalformedGetResponseTest, RequesterFutureFailsInsteadOfHanging) { } }); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port2 = + service2.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port2, 0); // When @@ -2242,8 +2270,9 @@ TEST(TransferServiceP2PTest, GetWritesToLocalDestinationNotToPeerNamedPath) { : MessageType::kError); }); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port2 = + service2.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port2, 0); // When @@ -2270,11 +2299,13 @@ TEST(TransferServiceP2PTest, GetReusesSinglePooledConnectionAfterSuccessAndError) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(0, /*threads=*/4, /*conn_pool_per_peer=*/1); + int port1 = service1.Initialize(CwdBaseDir(), 0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(0, /*threads=*/4, /*conn_pool_per_peer=*/1); + int port2 = service2.Initialize(CwdBaseDir(), 0, /*threads=*/4, + /*conn_pool_per_peer=*/1); ASSERT_GT(port2, 0); std::string peer1_addr = "127.0.0.1:" + std::to_string(port1); @@ -2315,11 +2346,11 @@ TEST(TransferServiceP2PTest, TEST(TransferServiceP2PTest, BidirectionalConcurrentGetAndPut) { // Given TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); ASSERT_GT(port1, 0); TransferService service2("127.0.0.1"); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); ASSERT_GT(port2, 0); std::string addr1 = "127.0.0.1:" + std::to_string(port1); @@ -2393,8 +2424,9 @@ TEST(TransferServiceP2PTest, int silent_listener = ListenOnLocalEphemeralPort(&silent_port); ASSERT_GE(silent_listener, 0); TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/2, - /*conn_pool_per_peer=*/1); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/2, + /*conn_pool_per_peer=*/1); ASSERT_GT(port1, 0); const std::string silent_addr = "127.0.0.1:" + std::to_string(silent_port); constexpr int kNumTasks = 6; // More than the 2 workers, so 4 stay queued. @@ -2420,5 +2452,272 @@ TEST(TransferServiceP2PTest, if (silent_listener >= 0) close(silent_listener); } +// A throwaway directory tree for the base-directory tests: +// // root(), canonical +// //base/ base(), the directory a service is confined to +// Removed with everything in it when the object goes out of scope. +class ScopedTestTree { + public: + ScopedTestTree() { + root_ = std::filesystem::temp_directory_path() / + ("transfer_service_base_dir_" + GenerateUuid()); + std::filesystem::create_directories(root_ / "base"); + root_ = std::filesystem::canonical(root_); + } + ~ScopedTestTree() { std::filesystem::remove_all(root_); } + + std::filesystem::path root() const { return root_; } + std::filesystem::path base() const { return root_ / "base"; } + + private: + std::filesystem::path root_; +}; + +// Sends one kPutObj with `payload` for `dest_obj_id` on `client_fd` and returns +// the type of the reply, or kGetObj (never a reply type) when none arrives. A +// kError reply must name the request's task; an ACK carries no task_id. +MessageType PutOverRawSocket(int client_fd, const std::string& task_id, + const std::string& dest_obj_id, + const std::string& payload) { + ObjInfoHeader request = BuildPutRequest(task_id, dest_obj_id, payload.size()); + EXPECT_TRUE(SendAll(client_fd, &request, kHeaderSize).ok()); + EXPECT_TRUE(SendAll(client_fd, payload.data(), payload.size()).ok()); + ObjInfoHeader response; + if (!RecvHeaderWithin(client_fd, &response, /*timeout_ms=*/10000)) { + return MessageType::kGetObj; + } + if (response.type == MessageType::kError) { + EXPECT_STREQ(response.task_id, task_id.c_str()); + } + return response.type; +} + +// A kPutObj whose destination resolves outside the service's base directory, +// whether by an absolute path or by "..", is answered with kError, writes +// nothing, and leaves the connection usable for a destination inside it. +TEST(TransferServiceP2PTest, PutOutsideBaseDirIsRejectedAndSocketStaysUsable) { + // Given + ScopedTestTree tree; + TransferService service1("127.0.0.1"); + int port1 = service1.Initialize(tree.base().native(), /*listen_port=*/0, + /*threads=*/4, /*conn_pool_per_peer=*/1); + ASSERT_GT(port1, 0); + const std::string payload(64 * 1024, 'P'); + const std::string absolute_outside = (tree.root() / "outside.obj").native(); + const std::string traversal_outside = + (tree.base() / ".." / "traversal.obj").native(); + const std::string inside = (tree.base() / "inside.obj").native(); + int client_fd = ConnectToLocalPort(port1); + ASSERT_GE(client_fd, 0); + + // When + const MessageType actual_absolute_reply = + PutOverRawSocket(client_fd, "absolute_task", absolute_outside, payload); + const MessageType actual_traversal_reply = + PutOverRawSocket(client_fd, "traversal_task", traversal_outside, payload); + const MessageType actual_inside_reply = + PutOverRawSocket(client_fd, "inside_task", inside, payload); + + // Then + EXPECT_EQ(MessageType::kError, actual_absolute_reply); + EXPECT_EQ(MessageType::kError, actual_traversal_reply); + EXPECT_EQ(MessageType::kAck, actual_inside_reply); + EXPECT_FALSE(std::filesystem::exists(absolute_outside)); + EXPECT_FALSE(std::filesystem::exists(absolute_outside + ".tmp")); + EXPECT_FALSE(std::filesystem::exists(tree.root() / "traversal.obj")); + EXPECT_FALSE(std::filesystem::exists(tree.root() / "traversal.obj.tmp")); + VerifyFileContentAndRemove(inside, payload); + + close(client_fd); + service1.Shutdown(); +} + +// A kGetObj for a file outside the base directory is answered with kError, +// whether it names the file directly, through "..", or through a symlink +// planted inside the base directory. The same socket still serves a file that +// is inside. +TEST(TransferServiceP2PTest, GetOutsideBaseDirIsRejectedAndSocketStaysUsable) { + // Given + ScopedTestTree tree; + TransferService service1("127.0.0.1"); + int port1 = service1.Initialize(tree.base().native(), /*listen_port=*/0, + /*threads=*/4, /*conn_pool_per_peer=*/1); + ASSERT_GT(port1, 0); + const std::string secret = (tree.root() / "secret.obj").native(); + WriteTestFile(secret, "outside the base directory"); + const std::string expected_inside_data = "inside the base directory"; + const std::string inside = (tree.base() / "inside.obj").native(); + WriteTestFile(inside, expected_inside_data); + const std::filesystem::path link = tree.base() / "link_to_secret.obj"; + std::filesystem::create_symlink(secret, link); + const std::string requester_addr = "127.0.0.1:1"; // Unused by the service. + int client_fd = ConnectToLocalPort(port1); + ASSERT_GE(client_fd, 0); + + // When + const ObjInfoHeader actual_absolute_reply = SendRequestAndRecvResponse( + client_fd, + BuildGetRequest("absolute_task", secret, "dest.obj", requester_addr)); + const ObjInfoHeader actual_traversal_reply = SendRequestAndRecvResponse( + client_fd, BuildGetRequest("traversal_task", + (tree.base() / ".." / "secret.obj").native(), + "dest.obj", requester_addr)); + const ObjInfoHeader actual_symlink_reply = SendRequestAndRecvResponse( + client_fd, BuildGetRequest("symlink_task", link.native(), "dest.obj", + requester_addr)); + const ObjInfoHeader actual_inside_reply = SendRequestAndRecvResponse( + client_fd, + BuildGetRequest("inside_task", inside, "dest.obj", requester_addr)); + const std::string actual_inside_data = + RecvGetPayloadAndAck(client_fd, actual_inside_reply); + + // Then + EXPECT_EQ(MessageType::kError, actual_absolute_reply.type); + EXPECT_STREQ(actual_absolute_reply.task_id, "absolute_task"); + EXPECT_EQ(MessageType::kError, actual_traversal_reply.type); + EXPECT_STREQ(actual_traversal_reply.task_id, "traversal_task"); + EXPECT_EQ(MessageType::kError, actual_symlink_reply.type); + EXPECT_STREQ(actual_symlink_reply.task_id, "symlink_task"); + EXPECT_EQ(MessageType::kRespondToGetObj, actual_inside_reply.type); + EXPECT_EQ(expected_inside_data, actual_inside_data); + + close(client_fd); + service1.Shutdown(); +} + +// An AsyncGet whose local destination is outside the base directory fails at +// once, with the reason, and nothing is sent to the peer. +TEST(TransferServiceP2PTest, + GetWithLocalDestinationOutsideBaseDirFailsBeforeContactingPeer) { + // Given + ScopedTestTree tree; + int peer_port = 0; + int peer_listener = ListenOnLocalEphemeralPort(&peer_port); + ASSERT_GE(peer_listener, 0); + TransferService service1("127.0.0.1"); + int port1 = service1.Initialize(tree.base().native(), /*listen_port=*/0, + /*threads=*/4, /*conn_pool_per_peer=*/1); + ASSERT_GT(port1, 0); + const std::string outside_dest = (tree.root() / "outside_dest.obj").native(); + + // When + std::future future = service1.AsyncGet( + "any_source_obj", "127.0.0.1:" + std::to_string(peer_port), outside_dest); + const bool actual_ready = FutureReadyWithin(future, std::chrono::seconds(5)); + const bool actual_peer_idle = ListenerStaysIdle(peer_listener, 300); + + // Then + ASSERT_TRUE(actual_ready) << "AsyncGet did not fail fast"; + EXPECT_THROW( + { + try { + future.get(); + } catch (const std::runtime_error& e) { + EXPECT_THAT(e.what(), + testing::HasSubstr("outside the base directory")); + throw; + } + }, + std::runtime_error); + EXPECT_TRUE(actual_peer_idle) << "a request was sent for a rejected Get"; + EXPECT_FALSE(std::filesystem::exists(outside_dest)); + + close(peer_listener); + service1.Shutdown(); +} + +// Relative object ids are resolved against the working directory and then +// confined like any other: with the working directory as the base, "../x" is +// rejected and "x" is served. Every other test in this file relies on this. +TEST(TransferServiceP2PTest, RelativeIdsAreConfinedToWorkingDirectoryBase) { + // Given + TransferService service1("127.0.0.1"); + int port1 = + service1.Initialize(CwdBaseDir(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); + ASSERT_GT(port1, 0); + const std::string payload = "stays in the working directory"; + const std::string escaped = "../cwd_escape_" + GenerateUuid() + ".obj"; + const std::string inside = "cwd_inside_" + GenerateUuid() + ".obj"; + int client_fd = ConnectToLocalPort(port1); + ASSERT_GE(client_fd, 0); + + // When + const MessageType actual_escaped_reply = + PutOverRawSocket(client_fd, "escaped_task", escaped, payload); + const MessageType actual_inside_reply = + PutOverRawSocket(client_fd, "inside_task", inside, payload); + + // Then + EXPECT_EQ(MessageType::kError, actual_escaped_reply); + EXPECT_FALSE(std::filesystem::exists(escaped)); + EXPECT_FALSE(std::filesystem::exists(escaped + ".tmp")); + EXPECT_EQ(MessageType::kAck, actual_inside_reply); + VerifyFileContentAndRemove(inside, payload); + + close(client_fd); + service1.Shutdown(); +} + +// The base directory is compared by its real location: a service configured +// with a symlink to the directory accepts ids spelled either way. +TEST(TransferServiceP2PTest, BaseDirGivenAsSymlinkIsResolved) { + // Given + ScopedTestTree tree; + const std::filesystem::path base_link = tree.root() / "base_link"; + std::filesystem::create_directory_symlink(tree.base(), base_link); + TransferService service1("127.0.0.1"); + int port1 = + service1.Initialize(base_link.native(), /*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1); + ASSERT_GT(port1, 0); + const std::string payload = "reachable through either spelling"; + const std::string via_real = (tree.base() / "via_real.obj").native(); + const std::string via_link = (base_link / "via_link.obj").native(); + int client_fd = ConnectToLocalPort(port1); + ASSERT_GE(client_fd, 0); + + // When + const MessageType actual_real_reply = + PutOverRawSocket(client_fd, "real_task", via_real, payload); + const MessageType actual_link_reply = + PutOverRawSocket(client_fd, "link_task", via_link, payload); + + // Then + EXPECT_EQ(MessageType::kAck, actual_real_reply); + EXPECT_EQ(MessageType::kAck, actual_link_reply); + VerifyFileContentAndRemove(via_real, payload); + VerifyFileContentAndRemove((tree.base() / "via_link.obj").native(), payload); + + close(client_fd); + service1.Shutdown(); +} + +// Initialize() creates a missing base directory and refuses one that exists +// but is not a directory. +TEST(TransferServiceP2PTest, InitializeCreatesBaseDirAndRejectsNonDirectory) { + // Given + ScopedTestTree tree; + const std::filesystem::path missing = tree.root() / "not" / "yet" / "there"; + const std::string regular_file = (tree.root() / "a_file").native(); + WriteTestFile(regular_file, "not a directory"); + TransferService service_missing("127.0.0.1"); + TransferService service_file("127.0.0.1"); + + // When + const int actual_missing_port = + service_missing.Initialize(missing.native(), /*listen_port=*/0, + /*threads=*/4, /*conn_pool_per_peer=*/1); + const int actual_file_port = service_file.Initialize( + regular_file, /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1); + + // Then + EXPECT_GT(actual_missing_port, 0); + EXPECT_TRUE(std::filesystem::is_directory(missing)); + EXPECT_EQ(-1, actual_file_port); + + service_missing.Shutdown(); +} + } // namespace } // namespace ml_flashpoint::replication::transfer_service diff --git a/tests/replication/transfer_service/transfer_service_test.cpp b/tests/replication/transfer_service/transfer_service_test.cpp index 774b2e8..ff581e9 100644 --- a/tests/replication/transfer_service/transfer_service_test.cpp +++ b/tests/replication/transfer_service/transfer_service_test.cpp @@ -14,6 +14,9 @@ #include "transfer_service.h" +#include +#include + #include "absl/log/log.h" #include "absl/strings/str_format.h" #include "gmock/gmock.h" @@ -22,6 +25,10 @@ namespace ml_flashpoint::replication::transfer_service { namespace { +// Base directory for the services in these tests. The working directory is +// what the tests write their object files to, by relative name. +std::string CwdBaseDir() { return std::filesystem::current_path().native(); } + TEST(TransferServiceTest, InitializeSucceedsWithFixedValidPorts) { constexpr int kNumThreads = 4; constexpr int kNumConnections = 10; @@ -29,47 +36,62 @@ TEST(TransferServiceTest, InitializeSucceedsWithFixedValidPorts) { TransferService service; // Use an ephemeral port (0) and let the OS assign an available port. - int actual_port = service.Initialize(0, kNumThreads, kNumConnections); + int actual_port = + service.Initialize(CwdBaseDir(), 0, kNumThreads, kNumConnections); EXPECT_GT(actual_port, 0) << "Initialize failed to return a valid ephemeral port."; } TEST(TransferServiceTest, InitializeFailsWithInvalidPort) { TransferService service; - EXPECT_DEATH(service.Initialize(-1, 4, 10), + EXPECT_DEATH(service.Initialize(CwdBaseDir(), -1, 4, 10), "listen_port must be non-negative"); } TEST(TransferServiceTest, InitializeFailsWithInvalidThreads) { TransferService service; - EXPECT_DEATH(service.Initialize(8080, 0, 10), "threads must be positive"); + EXPECT_DEATH(service.Initialize(CwdBaseDir(), 8080, 0, 10), + "threads must be positive"); } TEST(TransferServiceTest, InitializeFailsWithInvalidConnPool) { TransferService service; - EXPECT_DEATH(service.Initialize(8080, 4, 0), + EXPECT_DEATH(service.Initialize(CwdBaseDir(), 8080, 4, 0), "conn_pool_per_peer must be positive"); } +TEST(TransferServiceTest, InitializeFailsWithEmptyBaseDir) { + // Tests that a service is not started without a base directory to confine + // object ids to: there is no default such as the working directory. + // Given + TransferService service; + + // When + const int actual_port = service.Initialize(""); + + // Then + EXPECT_EQ(-1, actual_port); +} + TEST(TransferServiceTest, InitializeSucceedsWithEphemeralPort) { TransferService service; - EXPECT_GT(service.Initialize(0, 4, 10), 0); + EXPECT_GT(service.Initialize(CwdBaseDir(), 0, 4, 10), 0); } TEST(TransferServiceTest, DoubleInitialize) { TransferService service; - int port = service.Initialize(); + int port = service.Initialize(CwdBaseDir()); EXPECT_GT(port, 0); - EXPECT_EQ(service.Initialize(), port); + EXPECT_EQ(service.Initialize(CwdBaseDir()), port); } TEST(TransferServiceTest, InitializeTwoServices) { TransferService service1; TransferService service2; - int port1 = service1.Initialize(); + int port1 = service1.Initialize(CwdBaseDir()); LOG(INFO) << "port1: " << port1; EXPECT_GT(port1, 0); - int port2 = service2.Initialize(); + int port2 = service2.Initialize(CwdBaseDir()); LOG(INFO) << "port2: " << port2; EXPECT_GT(port2, 0); EXPECT_NE(port1, port2); @@ -77,7 +99,7 @@ TEST(TransferServiceTest, InitializeTwoServices) { TEST(TransferServiceTest, NormalShutdown) { TransferService service; - service.Initialize(); + service.Initialize(CwdBaseDir()); service.Shutdown(); // Should not crash. }