guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libutil: disallow Fd{Sink,Source} copy and move

Browse Source 
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
This commit is contained in:
eldritch horrors
2025-06-17 14:34:05 +02:00
parent 8835b2f057
commit fc18a6d170
10 changed files with 121 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/build/derivation-goal.cc
+7 -10
View File
@@ -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. */
lix/libstore/build/hook-instance.cc
+3 -3
View File
@@ -72,12 +72,12 @@ HookInstance::HookInstance()
toHook = std::move(toHook_.writeSide);
builderOut = std::move(builderOut_.readSide);
sink = FdSink(toHook.get());
sink = std::make_unique<FdSink>(toHook.get());
std::map<std::string, Config::SettingInfo> 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;
}
lix/libstore/build/hook-instance.hh
+1 -1
View File
@@ -29,7 +29,7 @@ struct HookInstance
*/
Pid pid;
FdSink sink;
std::unique_ptr<FdSink> sink;
std::map<ActivityId, Activity> activities;
lix/libstore/legacy-ssh-store.cc
+58 -78
View File
@@ -58,8 +58,8 @@ struct LegacySSHStore final : public Store
struct Connection
{
std::unique_ptr<SSH::Connection> sshConn;
FdSink to;
FdSource from;
std::unique_ptr<FdSink> to;
std::unique_ptr<FdSource> 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<FdSink>(conn->sshConn->socket.get());
conn->from = std::make_unique<FdSource>(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<AsyncGeneratorInputStream>([](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<BuildResult>::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<StorePathSet>::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<StorePathSet>::read(*conn);
} catch (...) {
lix/libstore/remote-store-connection.hh
+8 -8
View File
@@ -27,12 +27,12 @@ struct RemoteStore::Connection
/**
* Send with this.
*/
FdSink to;
std::unique_ptr<FdSink> to;
/**
* Receive with this.
*/
FdSource from;
std::unique_ptr<FdSource> 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<LastArgT>(); }) {
((handle->to << std::forward<Args>(args)), ...);
handle->to.flush();
if constexpr (requires { *handle->to << std::declval<LastArgT>(); }) {
((*handle->to << std::forward<Args>(args)), ...);
handle->to->flush();
LIX_TRY_AWAIT(processStderr());
} else {
using ImmediateArgsIdxs = std::make_index_sequence<sizeof...(Args) - 1>;
AllArgsT allArgs(std::forward<Args>(args)...);
[&]<size_t... Ids>(std::integer_sequence<size_t, Ids...>) {
((handle->to << std::forward<std::tuple_element_t<Ids, AllArgsT>>(
((*handle->to << std::forward<std::tuple_element_t<Ids, AllArgsT>>(
std::get<Ids>(allArgs)
)),
...);
handle->to.flush();
handle->to->flush();
}(ImmediateArgsIdxs{});
LIX_TRY_AWAIT(withFramedSinkAsync(std::get<LastArgIdx>(allArgs)));
lix/libstore/remote-store.cc
+30 -29
View File
@@ -38,7 +38,7 @@ RemoteStore::RemoteStore(const RemoteStoreConfig & config)
std::max(1, (int) config.maxConnections),
[this]() { return openAndInitConnection(); },
[this](const ref<Connection> & r) {
return r->to.good() && r->from.good()
return r->to->good() && r->from->good()
&& std::chrono::duration_cast<std::chrono::seconds>(
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<std::optional<TrustedFlag>>::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<AsyncGeneratorInputStream>([](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<Result<RemoteStore::Connection::RemoteError>> RemoteStore::Connectio
try {
while (true) {
auto msg = readNum<uint64_t>(from);
auto msg = readNum<uint64_t>(*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<ActivityId>(from);
auto lvl = (Verbosity) readInt(from);
auto type = (ActivityType) readInt(from);
auto s = readString(from);
auto fields = readFields(from);
auto parent = readNum<ActivityId>(from);
auto act = readNum<ActivityId>(*from);
auto lvl = (Verbosity) readInt(*from);
auto type = (ActivityType) readInt(*from);
auto s = readString(*from);
auto fields = readFields(*from);
auto parent = readNum<ActivityId>(*from);
logger->startActivity(act, lvl, type, s, fields, parent);
}
else if (msg == STDERR_STOP_ACTIVITY) {
auto act = readNum<ActivityId>(from);
auto act = readNum<ActivityId>(*from);
logger->stopActivity(act);
}
else if (msg == STDERR_RESULT) {
auto act = readNum<ActivityId>(from);
auto type = (ResultType) readInt(from);
auto fields = readFields(from);
auto act = readNum<ActivityId>(*from);
auto type = (ResultType) readInt(*from);
auto fields = readFields(*from);
logger->result(act, type, fields);
}
@@ -837,7 +838,7 @@ kj::Promise<Result<void>> 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();
}
lix/libstore/ssh-store.cc
+2 -2
View File
@@ -99,8 +99,8 @@ ref<RemoteStore::Connection> 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<FdSink>(conn->sshConn->socket.get());
conn->from = std::make_unique<FdSource>(conn->sshConn->socket.get());
return conn;
}
lix/libstore/uds-remote-store.cc
+2 -2
View File
@@ -60,8 +60,8 @@ ref<RemoteStore::Connection> 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<FdSource>(conn->fd.get());
conn->to = std::make_unique<FdSink>(conn->fd.get());
conn->startTime = std::chrono::steady_clock::now();
lix/libutil/serialise.hh
+8 -17
View File
@@ -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:
lix/nix/daemon.cc
+2 -2
View File
@@ -409,8 +409,8 @@ static void daemonLoop(AsyncIoRoot & aio, std::optional<TrustedFlag> 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) {