From fc18a6d170f16a4d5a192b1f85a6fa58abb7fbbd Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Mon, 16 Jun 2025 18:51:59 +0200 Subject: [PATCH] libutil: disallow Fd{Sink,Source} copy and move it was never safe. both discarded the buffer of the source object, possibly leading to silent data corruption. FdSource discarded the fancy EOF error string as well, possibly causing bad error reports Change-Id: Ib5c07986471b5af03d707230cd487259201952e9 --- lix/libstore/build/derivation-goal.cc | 17 ++- lix/libstore/build/hook-instance.cc | 6 +- lix/libstore/build/hook-instance.hh | 2 +- lix/libstore/legacy-ssh-store.cc | 136 ++++++++++-------------- lix/libstore/remote-store-connection.hh | 16 +-- lix/libstore/remote-store.cc | 59 +++++----- lix/libstore/ssh-store.cc | 4 +- lix/libstore/uds-remote-store.cc | 4 +- lix/libutil/serialise.hh | 25 ++--- lix/nix/daemon.cc | 4 +- 10 files changed, 121 insertions(+), 152 deletions(-) diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index 68dfda56f..6767bc37e 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1039,13 +1039,10 @@ HookReply DerivationGoal::tryBuildHook() try { /* Send the request to the hook. */ - worker.hook.instance->sink - << "try" - << (slotToken.valid() ? 1 : 0) - << drv->platform - << worker.store.printStorePath(drvPath) - << parsedDrv->getRequiredSystemFeatures(); - worker.hook.instance->sink.flush(); + *worker.hook.instance->sink << "try" << (slotToken.valid() ? 1 : 0) << drv->platform + << worker.store.printStorePath(drvPath) + << parsedDrv->getRequiredSystemFeatures(); + worker.hook.instance->sink->flush(); /* Read the first line of input, which should be a word indicating whether the hook wishes to perform the build. */ @@ -1108,7 +1105,7 @@ HookReply DerivationGoal::tryBuildHook() /* Tell the hook all the inputs that have to be copied to the remote system. */ - hook->sink << CommonProto::write({worker.store}, inputPaths); + *hook->sink << CommonProto::write({worker.store}, inputPaths); /* Tell the hooks the missing outputs that have to be copied back from the remote system. */ @@ -1119,10 +1116,10 @@ HookReply DerivationGoal::tryBuildHook() if (buildMode != bmCheck && status.known && status.known->isValid()) continue; missingOutputs.insert(outputName); } - hook->sink << CommonProto::write({worker.store}, missingOutputs); + *hook->sink << CommonProto::write({worker.store}, missingOutputs); } - hook->sink = FdSink(); + hook->sink = nullptr; hook->toHook.reset(); /* Create the log file and pipe. */ diff --git a/lix/libstore/build/hook-instance.cc b/lix/libstore/build/hook-instance.cc index c4e82919b..32adf893c 100644 --- a/lix/libstore/build/hook-instance.cc +++ b/lix/libstore/build/hook-instance.cc @@ -72,12 +72,12 @@ HookInstance::HookInstance() toHook = std::move(toHook_.writeSide); builderOut = std::move(builderOut_.readSide); - sink = FdSink(toHook.get()); + sink = std::make_unique(toHook.get()); std::map settings; globalConfig.getSettings(settings, true); for (auto & setting : settings) - sink << 1 << setting.first << setting.second.value; - sink << 0; + *sink << 1 << setting.first << setting.second.value; + *sink << 0; } diff --git a/lix/libstore/build/hook-instance.hh b/lix/libstore/build/hook-instance.hh index 38fd85ccb..12841fd4f 100644 --- a/lix/libstore/build/hook-instance.hh +++ b/lix/libstore/build/hook-instance.hh @@ -29,7 +29,7 @@ struct HookInstance */ Pid pid; - FdSink sink; + std::unique_ptr sink; std::map activities; diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index b5805b799..a1d3b9621 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -58,8 +58,8 @@ struct LegacySSHStore final : public Store struct Connection { std::unique_ptr sshConn; - FdSink to; - FdSource from; + std::unique_ptr to; + std::unique_ptr from; ServeProto::Version remoteVersion; Store * store = nullptr; bool good = true; @@ -74,8 +74,8 @@ struct LegacySSHStore final : public Store */ operator ServeProto::ReadConn () { - return ServeProto::ReadConn { - .from = from, + return ServeProto::ReadConn{ + .from = *from, .store = *store, .version = remoteVersion, }; @@ -136,18 +136,18 @@ struct LegacySSHStore final : public Store ? "" : " --store " + shellEscape(config_.remoteStore.get())) ); - conn->to = FdSink(conn->sshConn->socket.get()); - conn->from = FdSource(conn->sshConn->socket.get()); + conn->to = std::make_unique(conn->sshConn->socket.get()); + conn->from = std::make_unique(conn->sshConn->socket.get()); conn->store = this; try { - conn->to << SERVE_MAGIC_1 << SERVE_PROTOCOL_VERSION; - conn->to.flush(); + *conn->to << SERVE_MAGIC_1 << SERVE_PROTOCOL_VERSION; + conn->to->flush(); - uint64_t magic = readLongLong(conn->from); + uint64_t magic = readLongLong(*conn->from); if (magic != SERVE_MAGIC_2) throw Error("'nix-store --serve' protocol mismatch from '%s'", host); - conn->remoteVersion = readInt(conn->from); + conn->remoteVersion = readInt(*conn->from); if (GET_PROTOCOL_MAJOR(conn->remoteVersion) != 0x200) throw Error("unsupported 'nix-store --serve' protocol version on '%s'", host); @@ -175,10 +175,10 @@ struct LegacySSHStore final : public Store debug("querying remote host '%s' for info on '%s'", host, printStorePath(path)); - conn->to << ServeProto::Command::QueryPathInfos << PathSet{printStorePath(path)}; - conn->to.flush(); + *conn->to << ServeProto::Command::QueryPathInfos << PathSet{printStorePath(path)}; + conn->to->flush(); - auto p = readString(conn->from); + auto p = readString(*conn->from); if (p.empty()) co_return result::success(nullptr); auto path2 = parseStorePath(p); assert(path == path2); @@ -189,7 +189,7 @@ struct LegacySSHStore final : public Store if (info->narHash == Hash::dummy) throw Error("NAR hash is now mandatory"); - auto s = readString(conn->from); + auto s = readString(*conn->from); assert(s == ""); co_return info; @@ -206,51 +206,40 @@ struct LegacySSHStore final : public Store if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 5) { - conn->to - << ServeProto::Command::AddToStoreNar - << printStorePath(info.path) - << (info.deriver ? printStorePath(*info.deriver) : "") - << info.narHash.to_string(Base::Base16, false); - conn->to << ServeProto::write(*conn, info.references); - conn->to - << info.registrationTime - << info.narSize - << info.ultimate - << info.sigs - << renderContentAddress(info.ca); + *conn->to << ServeProto::Command::AddToStoreNar << printStorePath(info.path) + << (info.deriver ? printStorePath(*info.deriver) : "") + << info.narHash.to_string(Base::Base16, false); + *conn->to << ServeProto::write(*conn, info.references); + *conn->to << info.registrationTime << info.narSize << info.ultimate << info.sigs + << renderContentAddress(info.ca); try { - TRY_AWAIT(copyNAR(source)->drainInto(conn->to)); + TRY_AWAIT(copyNAR(source)->drainInto(*conn->to)); } catch (...) { conn->good = false; throw; } - conn->to.flush(); + conn->to->flush(); } else { - conn->to - << ServeProto::Command::ImportPaths - << 1; + *conn->to << ServeProto::Command::ImportPaths << 1; try { - TRY_AWAIT(copyNAR(source)->drainInto(conn->to)); + TRY_AWAIT(copyNAR(source)->drainInto(*conn->to)); } catch (...) { conn->good = false; throw; } - conn->to - << exportMagic - << printStorePath(info.path); - conn->to << ServeProto::write(*conn, info.references); - conn->to - << (info.deriver ? printStorePath(*info.deriver) : "") - << 0 - << 0; - conn->to.flush(); - + *conn->to << exportMagic << printStorePath(info.path); + *conn->to << ServeProto::write(*conn, info.references); + *conn->to << (info.deriver ? printStorePath(*info.deriver) : "") << 0 << 0; + conn->to->flush(); } - if (readInt(conn->from) != 1) - throw Error("failed to add path '%s' to remote host '%s'", printStorePath(info.path), host); + if (readInt(*conn->from) != 1) { + throw Error( + "failed to add path '%s' to remote host '%s'", printStorePath(info.path), host + ); + } co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -260,10 +249,10 @@ struct LegacySSHStore final : public Store try { auto conn(TRY_AWAIT(connections->get())); - conn->to << ServeProto::Command::DumpStorePath << printStorePath(path); - conn->to.flush(); + *conn->to << ServeProto::Command::DumpStorePath << printStorePath(path); + conn->to->flush(); co_return make_box_ptr([](auto conn) -> WireFormatGenerator { - co_yield copyNAR(conn->from); + co_yield copyNAR(*conn->from); }(std::move(conn))); } catch (...) { co_return result::current_exception(); @@ -302,19 +291,16 @@ private: void putBuildSettings(Connection & conn) { - conn.to - << settings.maxSilentTime - << settings.buildTimeout; - if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 2) - conn.to - << settings.maxLogSize; + *conn.to << settings.maxSilentTime << settings.buildTimeout; + if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 2) { + *conn.to << settings.maxLogSize; + } if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 3) - conn.to - << 0 // buildRepeat hasn't worked for ages anyway - << 0; + *conn.to << 0 // buildRepeat hasn't worked for ages anyway + << 0; if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 7) { - conn.to << ((int) settings.keepFailed); + *conn.to << ((int) settings.keepFailed); } } @@ -326,14 +312,12 @@ public: try { auto conn(TRY_AWAIT(connections->get())); - conn->to - << ServeProto::Command::BuildDerivation - << printStorePath(drvPath); - writeDerivation(conn->to, *this, drv); + *conn->to << ServeProto::Command::BuildDerivation << printStorePath(drvPath); + writeDerivation(*conn->to, *this, drv); putBuildSettings(*conn); - conn->to.flush(); + conn->to->flush(); co_return ServeProto::Serialise::read(*conn); } catch (...) { @@ -351,7 +335,7 @@ public: auto conn(TRY_AWAIT(connections->get())); - conn->to << ServeProto::Command::BuildPaths; + *conn->to << ServeProto::Command::BuildPaths; Strings ss; for (auto & p : drvPaths) { auto sOrDrvPath = StorePathWithOutputs::tryFromDerivedPath(p); @@ -367,17 +351,17 @@ public: }, }, sOrDrvPath); } - conn->to << ss; + *conn->to << ss; putBuildSettings(*conn); - conn->to.flush(); + conn->to->flush(); BuildResult result; - result.status = (BuildResult::Status) readInt(conn->from); + result.status = (BuildResult::Status) readInt(*conn->from); if (!result.success()) { - conn->from >> result.errorMsg; + *conn->from >> result.errorMsg; throw Error(result.status, result.errorMsg); } @@ -416,11 +400,9 @@ public: auto conn(TRY_AWAIT(connections->get())); - conn->to - << ServeProto::Command::QueryClosure - << includeOutputs; - conn->to << ServeProto::write(*conn, paths); - conn->to.flush(); + *conn->to << ServeProto::Command::QueryClosure << includeOutputs; + *conn->to << ServeProto::write(*conn, paths); + conn->to->flush(); for (auto & i : ServeProto::Serialise::read(*conn)) out.insert(i); @@ -434,12 +416,10 @@ public: try { auto conn(TRY_AWAIT(connections->get())); - conn->to - << ServeProto::Command::QueryValidPaths - << false // lock - << maybeSubstitute; - conn->to << ServeProto::write(*conn, paths); - conn->to.flush(); + *conn->to << ServeProto::Command::QueryValidPaths << false // lock + << maybeSubstitute; + *conn->to << ServeProto::write(*conn, paths); + conn->to->flush(); co_return ServeProto::Serialise::read(*conn); } catch (...) { diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index 986316ec9..8c730b014 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -27,12 +27,12 @@ struct RemoteStore::Connection /** * Send with this. */ - FdSink to; + std::unique_ptr to; /** * Receive with this. */ - FdSource from; + std::unique_ptr from; /** * The store this connection belongs to. @@ -81,7 +81,7 @@ struct RemoteStore::Connection */ operator WorkerProto::ReadConn () { - return WorkerProto::ReadConn{from, *store, daemonVersion}; + return WorkerProto::ReadConn{*from, *store, daemonVersion}; } /** @@ -160,20 +160,20 @@ struct RemoteStore::ConnectionHandle // not have a serializer we assume it's a callback for a subframe protocol // and serialize all *preceding* arguments normally before handing over to // the subframing layer (which is then responsible for any error handling) - if constexpr (requires { handle->to << std::declval(); }) { - ((handle->to << std::forward(args)), ...); - handle->to.flush(); + if constexpr (requires { *handle->to << std::declval(); }) { + ((*handle->to << std::forward(args)), ...); + handle->to->flush(); LIX_TRY_AWAIT(processStderr()); } else { using ImmediateArgsIdxs = std::make_index_sequence; AllArgsT allArgs(std::forward(args)...); [&](std::integer_sequence) { - ((handle->to << std::forward>( + ((*handle->to << std::forward>( std::get(allArgs) )), ...); - handle->to.flush(); + handle->to->flush(); }(ImmediateArgsIdxs{}); LIX_TRY_AWAIT(withFramedSinkAsync(std::get(allArgs))); diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index 808d3b6ac..9f634b0f8 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -38,7 +38,7 @@ RemoteStore::RemoteStore(const RemoteStoreConfig & config) std::max(1, (int) config.maxConnections), [this]() { return openAndInitConnection(); }, [this](const ref & r) { - return r->to.good() && r->from.good() + return r->to->good() && r->from->good() && std::chrono::duration_cast( std::chrono::steady_clock::now() - r->startTime ) @@ -81,28 +81,29 @@ try { /* Send the magic greeting, check for the reply. */ try { conn.store = this; - conn.from.specialEndOfFileError = "Nix daemon disconnected unexpectedly (maybe it crashed?)"; - conn.to << WORKER_MAGIC_1; - conn.to.flush(); + conn.from->specialEndOfFileError = + "Nix daemon disconnected unexpectedly (maybe it crashed?)"; + *conn.to << WORKER_MAGIC_1; + conn.to->flush(); - uint64_t magic = readLongLong(conn.from); + uint64_t magic = readLongLong(*conn.from); if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch"); - conn.from >> conn.daemonVersion; + *conn.from >> conn.daemonVersion; if (GET_PROTOCOL_MAJOR(conn.daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION)) throw Error("Nix daemon protocol version not supported"); if (GET_PROTOCOL_MINOR(conn.daemonVersion) < MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION) throw Error("the Nix daemon version is too old"); - conn.to << PROTOCOL_VERSION; + *conn.to << PROTOCOL_VERSION; // Obsolete CPU affinity. - conn.to << 0; + *conn.to << 0; - conn.to << false; // obsolete reserveSpace + *conn.to << false; // obsolete reserveSpace - conn.to.flush(); - conn.daemonNixVersion = readString(conn.from); + conn.to->flush(); + conn.daemonNixVersion = readString(*conn.from); conn.remoteTrustsUs = WorkerProto::Serialise>::read(conn); auto ex = TRY_AWAIT(conn.processStderr()); @@ -157,8 +158,8 @@ try { for (auto & i : overrides) command << i.first << i.second.value; - StringSource{command.s}.drainInto(conn.to); - conn.to.flush(); + StringSource{command.s}.drainInto(*conn.to); + conn.to->flush(); auto ex = TRY_AWAIT(conn.processStderr()); if (ex.e) { std::rethrow_exception(ex.e); @@ -715,7 +716,7 @@ try { RemoteStore::Connection::~Connection() { try { - to.flush(); + to->flush(); } catch (...) { ignoreExceptionInDestructor(); } @@ -726,7 +727,7 @@ try { auto conn(TRY_AWAIT(getConnection())); TRY_AWAIT(conn.sendCommand(WorkerProto::Op::NarFromPath, printStorePath(path))); co_return make_box_ptr([](auto conn) -> WireFormatGenerator { - co_yield copyNAR(conn->from); + co_yield copyNAR(*conn->from); }(std::move(conn))); } catch (...) { co_return result::current_exception(); @@ -758,34 +759,34 @@ kj::Promise> RemoteStore::Connectio try { while (true) { - auto msg = readNum(from); + auto msg = readNum(*from); if (msg == STDERR_ERROR) { - co_return RemoteError{std::make_exception_ptr(readError(from))}; + co_return RemoteError{std::make_exception_ptr(readError(*from))}; } else if (msg == STDERR_NEXT) - printError(chomp(readString(from))); + printError(chomp(readString(*from))); else if (msg == STDERR_START_ACTIVITY) { - auto act = readNum(from); - auto lvl = (Verbosity) readInt(from); - auto type = (ActivityType) readInt(from); - auto s = readString(from); - auto fields = readFields(from); - auto parent = readNum(from); + auto act = readNum(*from); + auto lvl = (Verbosity) readInt(*from); + auto type = (ActivityType) readInt(*from); + auto s = readString(*from); + auto fields = readFields(*from); + auto parent = readNum(*from); logger->startActivity(act, lvl, type, s, fields, parent); } else if (msg == STDERR_STOP_ACTIVITY) { - auto act = readNum(from); + auto act = readNum(*from); logger->stopActivity(act); } else if (msg == STDERR_RESULT) { - auto act = readNum(from); - auto type = (ResultType) readInt(from); - auto fields = readFields(from); + auto act = readNum(*from); + auto type = (ResultType) readInt(*from); + auto fields = readFields(*from); logger->result(act, type, fields); } @@ -837,7 +838,7 @@ kj::Promise> RemoteStore::ConnectionHandle::withFramedSinkAsync( try { { FramedSinkHandler handler{*this, *handlerThreads.lock()}; - FramedSink sink((*this)->to, handler.ex); + FramedSink sink(*(*this)->to, handler.ex); TRY_AWAIT(fun(sink)); sink.flush(); } diff --git a/lix/libstore/ssh-store.cc b/lix/libstore/ssh-store.cc index 3bb75dadb..b005becc2 100644 --- a/lix/libstore/ssh-store.cc +++ b/lix/libstore/ssh-store.cc @@ -99,8 +99,8 @@ ref SSHStore::openConnection() command += " --store " + shellEscape(config_.remoteStore.get()); conn->sshConn = ssh.startCommand(command); - conn->to = FdSink(conn->sshConn->socket.get()); - conn->from = FdSource(conn->sshConn->socket.get()); + conn->to = std::make_unique(conn->sshConn->socket.get()); + conn->from = std::make_unique(conn->sshConn->socket.get()); return conn; } diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index 142ccb0c5..1ff6d418b 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -60,8 +60,8 @@ ref UDSRemoteStore::openConnection() nix::connect(conn->fd.get(), path ? *path : settings.nixDaemonSocketFile); - conn->from.fd = conn->fd.get(); - conn->to.fd = conn->fd.get(); + conn->from = std::make_unique(conn->fd.get()); + conn->to = std::make_unique(conn->fd.get()); conn->startTime = std::chrono::steady_clock::now(); diff --git a/lix/libutil/serialise.hh b/lix/libutil/serialise.hh index 09e18997f..80fe952dc 100644 --- a/lix/libutil/serialise.hh +++ b/lix/libutil/serialise.hh @@ -125,15 +125,10 @@ struct FdSink : BufferedSink FdSink() : fd(-1) { } FdSink(int fd) : fd(fd) { } - FdSink(FdSink&&) = default; - - FdSink & operator=(FdSink && s) - { - flush(); - fd = s.fd; - s.fd = -1; - return *this; - } + FdSink(const FdSink &) = delete; + FdSink(FdSink &&) = delete; + FdSink & operator=(const FdSink &) = delete; + FdSink & operator=(FdSink &&) = delete; ~FdSink(); @@ -159,14 +154,10 @@ struct FdSource : BufferedSource FdSource() : fd(-1) { } FdSource(int fd) : fd(fd) { } - FdSource(FdSource &&) = default; - - FdSource & operator=(FdSource && s) - { - fd = s.fd; - s.fd = -1; - return *this; - } + FdSource(const FdSource &) = delete; + FdSource(FdSource &&) = delete; + FdSource & operator=(const FdSource &) = delete; + FdSource & operator=(FdSource &&) = delete; bool good() override; protected: diff --git a/lix/nix/daemon.cc b/lix/nix/daemon.cc index a163e0c4b..9eb6f8590 100644 --- a/lix/nix/daemon.cc +++ b/lix/nix/daemon.cc @@ -409,8 +409,8 @@ static void daemonLoop(AsyncIoRoot & aio, std::optional forceTrustC */ static void forwardStdioConnection(RemoteStore & store) { auto conn = store.openConnectionWrapper(); - int from = conn->from.fd; - int to = conn->to.fd; + int from = conn->from->fd; + int to = conn->to->fd; auto nfds = std::max(from, STDIN_FILENO) + 1; while (true) {