From 595d906090910aa823491a8e3abf3b7509ff7815 Mon Sep 17 00:00:00 2001 From: Husam Date: Tue, 13 Oct 2026 03:13:33 +0000 Subject: [PATCH 1/3] fix(replication): confine object ids to a base directory A peer that can reach the listener could name any path the process can read as the source_obj_id of a kGetObj, and any path it can write as the dest_obj_id of a kPutObj. The service now takes a base directory at Initialize() and rejects every object id that does not resolve into it. Resolution follows symlinks in the part of the path that exists and normalizes the rest, so "..", "." and symlinks planted inside the base directory cannot lead out of it. The containment check is component-wise, so "/base2/x" is not inside "/base". The resolved path is the one the service then opens, renames or creates, so what was checked is what is used. base_dir itself is created if missing and must be a directory. Enforcement points: - HandleGetObjRequest: the peer's source_obj_id; kError on rejection, the connection stays usable. - HandleDataReceive: dest_obj_id, the one place every write goes through, for a peer's Put and for the local half of a Get; the payload is drained and kError sent, as for the other early failures. - AsyncGet: the local dest_obj_id is checked before anything is sent, so a bad id fails the future at once instead of after a full transfer. The Python binding makes base_dir a required argument. In C++ an empty base_dir means the current working directory, which is what the existing tests rely on with their relative object ids. Tests: unit tests for ResolvePathWithinBaseDir (inside, not yet existing, dot segments, traversal, sibling prefix, base itself, empty, symlink file and directory escapes, symlink staying inside, relative ids, overlong component), and P2P tests for a rejected Put and Get over a raw socket (absolute, "..", symlink) with the socket still serving ids inside, a fast-failing AsyncGet that never contacts the peer, the working-directory default, a base directory given as a symlink, and Initialize() creating a missing base directory and refusing a regular file. --- .../replication/transfer_service/bindings.cpp | 25 +- .../transfer_service/transfer_helpers.cpp | 41 +++ .../transfer_service/transfer_helpers.h | 22 ++ .../transfer_service/transfer_service.cpp | 87 +++++- .../transfer_service/transfer_service.h | 24 +- .../transfer_helpers_test.cpp | 212 ++++++++++++++ .../transfer_service_p2p_test.cpp | 270 ++++++++++++++++++ 7 files changed, 656 insertions(+), 25 deletions(-) diff --git a/src/ml_flashpoint/replication/transfer_service/bindings.cpp b/src/ml_flashpoint/replication/transfer_service/bindings.cpp index 0031d09..67de0ce 100644 --- a/src/ml_flashpoint/replication/transfer_service/bindings.cpp +++ b/src/ml_flashpoint/replication/transfer_service/bindings.cpp @@ -35,13 +35,24 @@ 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& self, int listen_port, + const std::string& base_dir, int threads, int conn_pool_per_peer, + int global_rank) { + return self.Initialize(listen_port, threads, conn_pool_per_peer, + global_rank, base_dir); + }, + py::arg("listen_port"), py::arg("base_dir"), 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 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.") .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..73c5ad2 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" @@ -82,7 +83,8 @@ TransferService::~TransferService() { } int TransferService::Initialize(int listen_port, int threads, - int conn_pool_size_per_peer, int global_rank) { + int conn_pool_size_per_peer, int global_rank, + const std::string& base_dir) { 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. canonical() follows + // symlinks and needs the directory to exist, so object ids are compared + // against the directory's real location. + std::error_code ec; + std::filesystem::path base = base_dir.empty() + ? std::filesystem::current_path(ec) + : std::filesystem::path(base_dir); + if (!ec && !base_dir.empty()) { + std::filesystem::create_directories(base, ec); + } + if (!ec) { + base_dir_ = std::filesystem::canonical(base, 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..1d977b3 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 @@ -72,11 +73,19 @@ class TransferService final { // workers in total. // conn_pool_per_peer: The size of the connection pool for each peer. // global_rank: The global rank of the process. + // 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. The directory is created if it does not exist. Empty + // means the current working directory; relative object ids are + // resolved against the current working directory either way. // // 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); + int conn_pool_per_peer = 16, int global_rank = -1, + const std::string& base_dir = ""); // Shuts down the transfer service gracefully. void Shutdown(); @@ -88,7 +97,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 +110,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 +180,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/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..ce2b60e 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 { @@ -2420,5 +2421,274 @@ 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(/*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1, /*global_rank=*/-1, + tree.base().native()); + 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(/*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1, /*global_rank=*/-1, + tree.base().native()); + 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(/*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1, /*global_rank=*/-1, + tree.base().native()); + 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(); +} + +// Without an explicit base directory the service is confined to its current +// working directory, which is what every other test in this file relies on. +TEST(TransferServiceP2PTest, DefaultBaseDirIsCurrentWorkingDirectory) { + // Given + TransferService service1("127.0.0.1"); + int port1 = service1.Initialize(/*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(/*listen_port=*/0, /*threads=*/4, + /*conn_pool_per_peer=*/1, /*global_rank=*/-1, + base_link.native()); + 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( + /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1, + /*global_rank=*/-1, missing.native()); + const int actual_file_port = service_file.Initialize( + /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1, + /*global_rank=*/-1, regular_file); + + // 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 From 697c3b95a664f6183c79c8af2d5bcb358b0f5226 Mon Sep 17 00:00:00 2001 From: Husam Date: Tue, 13 Oct 2026 03:13:33 +0000 Subject: [PATCH 2/3] feat(replication): pass the ML Flashpoint base directory to the transfer service ReplicationManager.initialize() takes base_dir and hands it to the TransferService it creates, so replication can only read and write checkpoint objects under the ML Flashpoint base path. It is required unless a pre-configured replication_transfer_service is given, in which case that service's own base directory applies. The NeMo wrapper passes flashpoint_base_container. The transfer-service tests switch their working directory to tmp_path and use it as the base directory, so the relative object ids they use stay inside it and no test files are left in the repository. New tests cover a Put to a destination outside the receiver's base directory, a Get of a file outside the peer's base directory, a Get whose local destination is outside the base directory, the ValueError when base_dir is missing, and the pre-configured service path that needs no base_dir. --- docs/user-guide.md | 8 +- .../adapter/nemo/wrapper_util.py | 2 +- .../replication/replication_manager.py | 22 ++++- tests/replication/test_replication_manager.py | 39 +++++++- .../test_replication_manager_e2e.py | 18 ++-- tests/replication/test_transer_service.py | 90 ++++++++++++++++++- 6 files changed, 163 insertions(+), 16 deletions(-) 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..7983e48 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 @@ -222,6 +241,7 @@ def initialize( bound_listen_port = self._transfer_service.initialize( listen_port, global_rank=dist.get_rank(), + base_dir=os.fspath(base_dir), ) if bound_listen_port <= 0: _LOGGER.error("Failed to initialize C++ TransferService") diff --git a/tests/replication/test_replication_manager.py b/tests/replication/test_replication_manager.py index eb9e3dc..758d142 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(0, global_rank=0, base_dir="/flashpoint/base") 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..ef77818 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(0, global_rank=rank, base_dir=base_dir) 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..bfe1229 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(listen_port=port1, base_dir=str(tmp_path)) + service2.initialize(listen_port=port2, base_dir=str(tmp_path)) 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, From faf6b147cc34d76c91fb4bdc35a97a29f3e4ea8f Mon Sep 17 00:00:00 2001 From: Husam Date: Tue, 13 Oct 2026 14:45:08 +0000 Subject: [PATCH 3/3] fix(replication): require an explicit base directory for the transfer service The previous commits let an empty base_dir mean the current working directory. That is one more implicit way to end up with a service whose reach depends on where the process happened to be started. base_dir is now the first parameter of Initialize(), has no default, and an empty value makes Initialize() return -1 like any other unusable base directory. The Python binding is bound directly to the C++ method, so it reads initialize(base_dir, listen_port=0, threads=16, conn_pool_per_peer=16, global_rank=-1). A caller that still passes the port first gets a TypeError instead of an unconfined service. The C++ tests pass their working directory explicitly (CwdBaseDir()), which is where they create their object files by relative name. The test that covered the working-directory default now covers relative ids with the working directory as the base; a new unit test covers the empty base_dir. --- .../replication/replication_manager.py | 4 +- .../replication/transfer_service/bindings.cpp | 21 +- .../transfer_service/transfer_service.cpp | 26 +- .../transfer_service/transfer_service.h | 23 +- tests/replication/test_replication_manager.py | 2 +- .../test_replication_manager_e2e.py | 2 +- tests/replication/test_transer_service.py | 4 +- .../transfer_service_p2p_test.cpp | 277 ++++++++++-------- .../transfer_service_test.cpp | 42 ++- 9 files changed, 224 insertions(+), 177 deletions(-) diff --git a/src/ml_flashpoint/replication/replication_manager.py b/src/ml_flashpoint/replication/replication_manager.py index 7983e48..9190b90 100644 --- a/src/ml_flashpoint/replication/replication_manager.py +++ b/src/ml_flashpoint/replication/replication_manager.py @@ -239,9 +239,9 @@ 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(), - base_dir=os.fspath(base_dir), ) if bound_listen_port <= 0: _LOGGER.error("Failed to initialize C++ TransferService") diff --git a/src/ml_flashpoint/replication/transfer_service/bindings.cpp b/src/ml_flashpoint/replication/transfer_service/bindings.cpp index 67de0ce..59fed50 100644 --- a/src/ml_flashpoint/replication/transfer_service/bindings.cpp +++ b/src/ml_flashpoint/replication/transfer_service/bindings.cpp @@ -36,23 +36,18 @@ PYBIND11_MODULE(transfer_service_ext, m) { // Synchronous methods remain unchanged (still blocking) .def( - "initialize", - [](TransferService& self, int listen_port, - const std::string& base_dir, int threads, int conn_pool_per_peer, - int global_rank) { - return self.Initialize(listen_port, threads, conn_pool_per_peer, - global_rank, base_dir); - }, - py::arg("listen_port"), py::arg("base_dir"), py::arg("threads") = 16, + "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 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.") + "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_service.cpp b/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp index 73c5ad2..7e90853 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp +++ b/src/ml_flashpoint/replication/transfer_service/transfer_service.cpp @@ -82,9 +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, - const std::string& base_dir) { +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."; @@ -101,18 +101,18 @@ 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. canonical() follows - // symlinks and needs the directory to exist, so object ids are compared - // against the directory's real location. - std::error_code ec; - std::filesystem::path base = base_dir.empty() - ? std::filesystem::current_path(ec) - : std::filesystem::path(base_dir); - if (!ec && !base_dir.empty()) { - std::filesystem::create_directories(base, ec); + // 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, 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); diff --git a/src/ml_flashpoint/replication/transfer_service/transfer_service.h b/src/ml_flashpoint/replication/transfer_service/transfer_service.h index 1d977b3..f2e7b29 100644 --- a/src/ml_flashpoint/replication/transfer_service/transfer_service.h +++ b/src/ml_flashpoint/replication/transfer_service/transfer_service.h @@ -67,25 +67,26 @@ 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 // workers in total. // conn_pool_per_peer: The size of the connection pool for each peer. // global_rank: The global rank of the process. - // 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. The directory is created if it does not exist. Empty - // means the current working directory; relative object ids are - // resolved against the current working directory either way. // // 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, - const std::string& base_dir = ""); + // 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(); diff --git a/tests/replication/test_replication_manager.py b/tests/replication/test_replication_manager.py index 758d142..b63e873 100644 --- a/tests/replication/test_replication_manager.py +++ b/tests/replication/test_replication_manager.py @@ -59,7 +59,7 @@ def mock_gather(): # Then assert replication_manager._listen_port == 12345 - mock_transfer_service_instance.initialize.assert_called_once_with(0, global_rank=0, base_dir="/flashpoint/base") + 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): diff --git a/tests/replication/test_replication_manager_e2e.py b/tests/replication/test_replication_manager_e2e.py index ef77818..6231417 100644 --- a/tests/replication/test_replication_manager_e2e.py +++ b/tests/replication/test_replication_manager_e2e.py @@ -27,7 +27,7 @@ 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, base_dir=base_dir) + port = service.initialize(base_dir, global_rank=rank) port_container.append(port) diff --git a/tests/replication/test_transer_service.py b/tests/replication/test_transer_service.py index bfe1229..3dfab1e 100644 --- a/tests/replication/test_transer_service.py +++ b/tests/replication/test_transer_service.py @@ -89,8 +89,8 @@ def fixture_transfer_services( addr1 = f"127.0.0.1:{port1}" addr2 = f"127.0.0.1:{port2}" - service1.initialize(listen_port=port1, base_dir=str(tmp_path)) - service2.initialize(listen_port=port2, base_dir=str(tmp_path)) + service1.initialize(str(tmp_path), listen_port=port1) + service2.initialize(str(tmp_path), listen_port=port2) yield service1, service2, addr1, addr2 diff --git a/tests/replication/transfer_service/transfer_service_p2p_test.cpp b/tests/replication/transfer_service/transfer_service_p2p_test.cpp index ce2b60e..1358215 100644 --- a/tests/replication/transfer_service/transfer_service_p2p_test.cpp +++ b/tests/replication/transfer_service/transfer_service_p2p_test.cpp @@ -112,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!"; @@ -138,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) @@ -168,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 @@ -210,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 @@ -268,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 = ""; @@ -304,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"; @@ -330,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; @@ -371,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!"; @@ -403,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; @@ -443,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; @@ -493,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"; @@ -529,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."; @@ -563,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."; @@ -606,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"; @@ -632,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"; @@ -659,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"; @@ -688,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"; @@ -722,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"; @@ -753,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"; @@ -785,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"; @@ -817,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"; @@ -859,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 @@ -1154,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"; @@ -1228,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"; @@ -1280,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"; @@ -1315,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"; @@ -1363,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()); @@ -1397,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); @@ -1421,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); @@ -1453,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); @@ -1519,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); @@ -1576,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"; @@ -1606,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()); } @@ -1617,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'); @@ -1652,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"; @@ -1684,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; @@ -1732,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); @@ -1789,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'); @@ -1838,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"; @@ -1885,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"; @@ -1931,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"); @@ -1972,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"; @@ -2011,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 @@ -2066,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'); @@ -2112,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"; @@ -2186,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 @@ -2243,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 @@ -2271,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); @@ -2316,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); @@ -2394,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. @@ -2468,9 +2499,8 @@ TEST(TransferServiceP2PTest, PutOutsideBaseDirIsRejectedAndSocketStaysUsable) { // Given ScopedTestTree tree; TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1, /*global_rank=*/-1, - tree.base().native()); + 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(); @@ -2510,9 +2540,8 @@ TEST(TransferServiceP2PTest, GetOutsideBaseDirIsRejectedAndSocketStaysUsable) { // Given ScopedTestTree tree; TransferService service1("127.0.0.1"); - int port1 = service1.Initialize(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1, /*global_rank=*/-1, - tree.base().native()); + 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"); @@ -2566,9 +2595,8 @@ TEST(TransferServiceP2PTest, 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=*/4, - /*conn_pool_per_peer=*/1, /*global_rank=*/-1, - tree.base().native()); + 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(); @@ -2598,13 +2626,15 @@ TEST(TransferServiceP2PTest, service1.Shutdown(); } -// Without an explicit base directory the service is confined to its current -// working directory, which is what every other test in this file relies on. -TEST(TransferServiceP2PTest, DefaultBaseDirIsCurrentWorkingDirectory) { +// 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(/*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 payload = "stays in the working directory"; const std::string escaped = "../cwd_escape_" + GenerateUuid() + ".obj"; @@ -2637,9 +2667,9 @@ TEST(TransferServiceP2PTest, BaseDirGivenAsSymlinkIsResolved) { 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(/*listen_port=*/0, /*threads=*/4, - /*conn_pool_per_peer=*/1, /*global_rank=*/-1, - base_link.native()); + 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(); @@ -2675,12 +2705,11 @@ TEST(TransferServiceP2PTest, InitializeCreatesBaseDirAndRejectsNonDirectory) { TransferService service_file("127.0.0.1"); // When - const int actual_missing_port = service_missing.Initialize( - /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1, - /*global_rank=*/-1, missing.native()); + 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( - /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1, - /*global_rank=*/-1, regular_file); + regular_file, /*listen_port=*/0, /*threads=*/4, /*conn_pool_per_peer=*/1); // Then EXPECT_GT(actual_missing_port, 0); 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. }