mirror of
https://github.com/microsoft/PowerToys.git
synced 2026-08-29 10:09:43 +02:00
Harden IPC pipe ownership and shutdown lifecycle (#48902)
## Summary of the Pull Request The two-way named-pipe IPC server (`TwoWayPipeMessageIPC`, shared by the runner, Settings, and Quick Access host) created every pipe instance without `FILE_FLAG_FIRST_PIPE_INSTANCE`. If a pipe with the same name already existed — for example a leftover instance from a previous run or another process — `CreateNamedPipe` would quietly create an *additional* instance and share the name instead of owning it. This makes `start_named_pipe_server` create the **first** instance with `FILE_FLAG_FIRST_PIPE_INSTANCE`, so `CreateNamedPipe` fails fast on a name collision and the server is the authoritative owner of its pipe name. ## PR Checklist - [ ] **Communication:** I've discussed this with core contributors already. If the work hasn't been agreed, this work might be rejected - [x] **Tests:** Added/updated and all pass - [x] **Localization:** All end-user-facing strings can be localized (N/A — no user-facing strings) - [x] **Dev docs:** Added/updated (N/A) - [x] **New binaries:** Added on the required places (N/A — no new binaries) ## Detailed Description of the Pull Request / Additional comments - The flag is applied **only** to the first instance. Subsequent instances continue to omit it, so the existing `PIPE_UNLIMITED_INSTANCES` behavior is fully preserved. - The change is contained to a single function in `src/common/interop/two_way_pipe_message_ipc.cpp`. Public signatures and the `PowerToys.Interop` ABI are unchanged, so the runner, Settings, and Quick Access host all benefit without any code changes on their side. ## Validation Steps Performed - The existing `Common.Interop.UnitTests` `TestSend` exercises the modified first-instance code path (`Start()` → `start_named_pipe_server`) and continues to pass — a full IPC round-trip still works. - Verified the updated `CreateNamedPipe` open-mode logic compiles cleanly. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 49797c8c-784d-47e6-bc0f-53464eecec4b
This commit is contained in:
Gordon Lam
1242
src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp
Normal file
1242
src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp
Normal file
File diff suppressed because it is too large
Load Diff
@@ -28,7 +28,7 @@
|
||||
<ClCompile>
|
||||
<AdditionalIncludeDirectories>..\;..\utils;..\Telemetry;..\..\;..\..\..\deps\;..\..\..\packages\Microsoft.Windows.ImplementationLibrary.1.0.260126.7\include;$(VCInstallDir)UnitTest\include;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories>
|
||||
<LanguageStandard>stdcpp23</LanguageStandard>
|
||||
<PreprocessorDefinitions>SPDLOG_WCHAR_TO_UTF8_SUPPORT;SPDLOG_HEADER_ONLY;%(PreprocessorDefinitions)</PreprocessorDefinitions>
|
||||
<PreprocessorDefinitions>SPDLOG_WCHAR_TO_UTF8_SUPPORT;SPDLOG_HEADER_ONLY;TWO_WAY_PIPE_MESSAGE_IPC_TESTS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
|
||||
</ClCompile>
|
||||
<Link>
|
||||
<AdditionalLibraryDirectories>$(VCInstallDir)UnitTest\lib;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories>
|
||||
@@ -73,6 +73,10 @@
|
||||
<ClCompile Include="Resources.Tests.cpp" />
|
||||
<ClCompile Include="TestStubs.cpp" />
|
||||
<ClCompile Include="UnhandledException.Tests.cpp" />
|
||||
<ClCompile Include="TwoWayPipeMessageIPC.Tests.cpp" />
|
||||
<ClCompile Include="..\interop\two_way_pipe_message_ipc.cpp">
|
||||
<PrecompiledHeader>NotUsing</PrecompiledHeader>
|
||||
</ClCompile>
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ClInclude Include="pch.h" />
|
||||
|
||||
@@ -120,6 +120,9 @@
|
||||
<ClCompile Include="UnhandledException.Tests.cpp">
|
||||
<Filter>Source Files\Integration</Filter>
|
||||
</ClCompile>
|
||||
<ClCompile Include="TwoWayPipeMessageIPC.Tests.cpp">
|
||||
<Filter>Source Files\Integration</Filter>
|
||||
</ClCompile>
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ClInclude Include="pch.h">
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,6 +1,28 @@
|
||||
#pragma once
|
||||
#include <functional>
|
||||
#include "pipe_caller_auth.h"
|
||||
|
||||
namespace two_way_pipe_message_ipc
|
||||
{
|
||||
// Outbound clients must never grant a server an impersonation-capable token.
|
||||
inline constexpr DWORD ClientOpenFlags = FILE_FLAG_OVERLAPPED | SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION;
|
||||
}
|
||||
|
||||
#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS
|
||||
namespace two_way_pipe_message_ipc_test
|
||||
{
|
||||
void FailThreadStartAfter(int successful_starts);
|
||||
void SetWaitNamedPipeEnteredEvent(HANDLE event);
|
||||
void SetHandlerCompletionEvents(HANDLE completed_event, HANDLE allow_return_event);
|
||||
void SetBeforeReplacementListenerEvents(HANDLE reached_event, HANDLE allow_creation_event);
|
||||
void SetAfterReplacementListenerEvents(HANDLE reached_event, HANDLE allow_continue_event);
|
||||
void FailHandlerThreadStartAfter(int successful_starts);
|
||||
void SetHandlerThreadStartAttemptEvent(HANDLE event);
|
||||
void SetOutputWritePendingEvent(HANDLE event);
|
||||
void ResetFaultInjection();
|
||||
}
|
||||
#endif
|
||||
|
||||
class TwoWayPipeMessageIPC
|
||||
{
|
||||
public:
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
#pragma once
|
||||
#include <atomic>
|
||||
#include <condition_variable>
|
||||
#include <memory>
|
||||
#include <utility>
|
||||
#include <Windows.h>
|
||||
#include "async_message_queue.h"
|
||||
#include <WinSafer.h>
|
||||
#include <accctrl.h>
|
||||
#include <aclapi.h>
|
||||
#include <list>
|
||||
#include <vector>
|
||||
#include "two_way_pipe_message_ipc.h"
|
||||
|
||||
class TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl
|
||||
@@ -17,6 +21,103 @@ public:
|
||||
void end();
|
||||
|
||||
private:
|
||||
struct OwnedPipeHandle
|
||||
{
|
||||
OwnedPipeHandle() = default;
|
||||
explicit OwnedPipeHandle(HANDLE handle) :
|
||||
handle(handle)
|
||||
{
|
||||
}
|
||||
~OwnedPipeHandle()
|
||||
{
|
||||
reset();
|
||||
}
|
||||
|
||||
OwnedPipeHandle(const OwnedPipeHandle&) = delete;
|
||||
OwnedPipeHandle& operator=(const OwnedPipeHandle&) = delete;
|
||||
|
||||
OwnedPipeHandle(OwnedPipeHandle&& other) noexcept :
|
||||
handle(other.release())
|
||||
{
|
||||
}
|
||||
OwnedPipeHandle& operator=(OwnedPipeHandle&& other) noexcept
|
||||
{
|
||||
if (this != &other)
|
||||
{
|
||||
reset(other.release());
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
|
||||
[[nodiscard]] HANDLE get() const
|
||||
{
|
||||
return handle;
|
||||
}
|
||||
[[nodiscard]] bool valid() const
|
||||
{
|
||||
return handle != INVALID_HANDLE_VALUE;
|
||||
}
|
||||
HANDLE release()
|
||||
{
|
||||
return std::exchange(handle, INVALID_HANDLE_VALUE);
|
||||
}
|
||||
void reset(HANDLE new_handle = INVALID_HANDLE_VALUE)
|
||||
{
|
||||
if (handle != INVALID_HANDLE_VALUE)
|
||||
{
|
||||
CloseHandle(handle);
|
||||
}
|
||||
handle = new_handle;
|
||||
}
|
||||
|
||||
private:
|
||||
HANDLE handle = INVALID_HANDLE_VALUE;
|
||||
};
|
||||
|
||||
struct ConnectionHandler
|
||||
{
|
||||
explicit ConnectionHandler(OwnedPipeHandle&& pipe) :
|
||||
pipe_handle(std::move(pipe))
|
||||
{
|
||||
}
|
||||
|
||||
OwnedPipeHandle pipe_handle;
|
||||
std::thread thread;
|
||||
bool completed = false;
|
||||
};
|
||||
|
||||
enum class LifecycleState
|
||||
{
|
||||
NotStarted,
|
||||
Starting,
|
||||
Running,
|
||||
Stopping,
|
||||
Stopped,
|
||||
};
|
||||
|
||||
struct PipeSecurityAttributes
|
||||
{
|
||||
SECURITY_DESCRIPTOR security_descriptor{};
|
||||
SECURITY_ATTRIBUTES attributes{};
|
||||
PACL dacl = nullptr;
|
||||
PSID logon_sid = nullptr;
|
||||
BYTE administrators_sid[SECURITY_MAX_SID_SIZE]{};
|
||||
BYTE local_system_sid[SECURITY_MAX_SID_SIZE]{};
|
||||
BYTE server_sid[SECURITY_MAX_SID_SIZE]{};
|
||||
|
||||
~PipeSecurityAttributes()
|
||||
{
|
||||
if (dacl)
|
||||
{
|
||||
LocalFree(dacl);
|
||||
}
|
||||
if (logon_sid)
|
||||
{
|
||||
HeapFree(GetProcessHeap(), 0, logon_sid);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
AsyncMessageQueue input_queue;
|
||||
AsyncMessageQueue output_queue;
|
||||
std::wstring output_pipe_name;
|
||||
@@ -24,11 +125,19 @@ private:
|
||||
std::thread input_queue_thread;
|
||||
std::thread output_queue_thread;
|
||||
std::thread input_pipe_thread;
|
||||
std::mutex lifecycle_mutex;
|
||||
std::condition_variable lifecycle_stopped;
|
||||
LifecycleState lifecycle_state = LifecycleState::NotStarted;
|
||||
std::mutex pipe_connect_handle_mutex; // For manipulating the current_connect_pipe
|
||||
std::mutex output_pipe_mutex;
|
||||
std::mutex connection_handlers_mutex;
|
||||
std::vector<std::shared_ptr<ConnectionHandler>> connection_handlers;
|
||||
std::wstring outgoing_message; // Store the updated json settings.
|
||||
|
||||
HANDLE current_connect_pipe_handle = NULL;
|
||||
bool closed = false;
|
||||
HANDLE active_output_pipe_handle = INVALID_HANDLE_VALUE;
|
||||
HANDLE pipe_security_token = nullptr;
|
||||
std::atomic_bool closed = false;
|
||||
TwoWayPipeMessageIPC::callback_function dispatch_inc_message_function;
|
||||
interop_auth::CallerPolicy caller_policy;
|
||||
interop_auth::VerificationCache caller_cache;
|
||||
@@ -37,9 +146,16 @@ private:
|
||||
void consume_output_queue_thread();
|
||||
BOOL GetLogonSID(HANDLE hToken, PSID* ppsid);
|
||||
VOID FreeLogonSID(PSID* ppsid);
|
||||
int change_pipe_security_allow_restricted_token(HANDLE handle, HANDLE token);
|
||||
bool create_pipe_security_attributes(HANDLE token, PipeSecurityAttributes& security_attributes);
|
||||
void start_threads(HANDLE token);
|
||||
void stop_started_threads();
|
||||
void cancel_active_output_io();
|
||||
HANDLE create_medium_integrity_token();
|
||||
void handle_pipe_connection(HANDLE input_pipe_handle);
|
||||
void handle_pipe_connection(const std::shared_ptr<ConnectionHandler>& handler);
|
||||
void finish_connection_handler(const std::shared_ptr<ConnectionHandler>& handler);
|
||||
bool start_connection_handler(OwnedPipeHandle&& pipe_handle);
|
||||
void reap_finished_connection_handlers();
|
||||
void cancel_and_wait_for_connection_handlers();
|
||||
void start_named_pipe_server(HANDLE token);
|
||||
void consume_input_queue_thread();
|
||||
};
|
||||
|
||||
@@ -2,9 +2,92 @@
|
||||
|
||||
#include <common/interop/two_way_pipe_message_ipc_impl.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <iterator>
|
||||
#include <system_error>
|
||||
|
||||
constexpr DWORD BUFSIZE = 1024;
|
||||
constexpr DWORD PipeClientAccess = FILE_READ_DATA |
|
||||
FILE_READ_ATTRIBUTES |
|
||||
READ_CONTROL |
|
||||
FILE_WRITE_DATA |
|
||||
FILE_WRITE_ATTRIBUTES |
|
||||
SYNCHRONIZE;
|
||||
constexpr DWORD PipeWaitIntervalMs = 100;
|
||||
|
||||
namespace
|
||||
{
|
||||
HANDLE duplicate_pipe_security_token(HANDLE token)
|
||||
{
|
||||
if (!token)
|
||||
{
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
HANDLE duplicate = nullptr;
|
||||
if (!DuplicateHandle(GetCurrentProcess(),
|
||||
token,
|
||||
GetCurrentProcess(),
|
||||
&duplicate,
|
||||
0,
|
||||
FALSE,
|
||||
DUPLICATE_SAME_ACCESS))
|
||||
{
|
||||
throw std::system_error(GetLastError(), std::system_category());
|
||||
}
|
||||
return duplicate;
|
||||
}
|
||||
}
|
||||
|
||||
#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS
|
||||
namespace
|
||||
{
|
||||
std::atomic_int thread_start_failure_after{ -1 };
|
||||
std::atomic<HANDLE> wait_named_pipe_entered_event{ nullptr };
|
||||
|
||||
void inject_thread_start_failure()
|
||||
{
|
||||
int remaining = thread_start_failure_after.load();
|
||||
while (remaining >= 0)
|
||||
{
|
||||
if (remaining == 0)
|
||||
{
|
||||
throw std::system_error(std::make_error_code(std::errc::resource_unavailable_try_again));
|
||||
}
|
||||
if (thread_start_failure_after.compare_exchange_weak(remaining, remaining - 1))
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
namespace two_way_pipe_message_ipc_test
|
||||
{
|
||||
void FailThreadStartAfter(int successful_starts)
|
||||
{
|
||||
thread_start_failure_after.store(successful_starts);
|
||||
}
|
||||
|
||||
void SetWaitNamedPipeEnteredEvent(HANDLE event)
|
||||
{
|
||||
wait_named_pipe_entered_event.store(event);
|
||||
}
|
||||
|
||||
void ResetFaultInjection()
|
||||
{
|
||||
thread_start_failure_after.store(-1);
|
||||
wait_named_pipe_entered_event.store(nullptr);
|
||||
}
|
||||
}
|
||||
#else
|
||||
namespace
|
||||
{
|
||||
void inject_thread_start_failure()
|
||||
{
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
TwoWayPipeMessageIPC::TwoWayPipeMessageIPC(
|
||||
std::wstring _input_pipe_name,
|
||||
@@ -19,6 +102,7 @@ TwoWayPipeMessageIPC::TwoWayPipeMessageIPC(
|
||||
|
||||
TwoWayPipeMessageIPC::~TwoWayPipeMessageIPC()
|
||||
{
|
||||
impl->end();
|
||||
delete impl;
|
||||
}
|
||||
|
||||
@@ -54,54 +138,136 @@ void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::send(std::wstring msg)
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::start(HANDLE _restricted_pipe_token)
|
||||
{
|
||||
output_queue_thread = std::thread(&TwoWayPipeMessageIPCImpl::consume_output_queue_thread, this);
|
||||
input_queue_thread = std::thread(&TwoWayPipeMessageIPCImpl::consume_input_queue_thread, this);
|
||||
input_pipe_thread = std::thread(&TwoWayPipeMessageIPCImpl::start_named_pipe_server, this, _restricted_pipe_token);
|
||||
std::scoped_lock lock(lifecycle_mutex);
|
||||
if (lifecycle_state != LifecycleState::NotStarted)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
caller_policy = {};
|
||||
pipe_security_token = duplicate_pipe_security_token(_restricted_pipe_token);
|
||||
closed.store(false);
|
||||
lifecycle_state = LifecycleState::Starting;
|
||||
try
|
||||
{
|
||||
start_threads(pipe_security_token);
|
||||
lifecycle_state = LifecycleState::Running;
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
closed.store(true);
|
||||
stop_started_threads();
|
||||
lifecycle_state = LifecycleState::Stopped;
|
||||
lifecycle_stopped.notify_all();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::end()
|
||||
{
|
||||
closed = true;
|
||||
input_queue.interrupt();
|
||||
input_queue_thread.join();
|
||||
output_queue.interrupt();
|
||||
output_queue_thread.join();
|
||||
pipe_connect_handle_mutex.lock();
|
||||
if (current_connect_pipe_handle != NULL)
|
||||
{
|
||||
//Cancels the Pipe currently waiting for a connection.
|
||||
CancelIoEx(current_connect_pipe_handle, NULL);
|
||||
std::unique_lock lock(lifecycle_mutex);
|
||||
if (lifecycle_state == LifecycleState::NotStarted || lifecycle_state == LifecycleState::Stopped)
|
||||
{
|
||||
lifecycle_state = LifecycleState::Stopped;
|
||||
return;
|
||||
}
|
||||
if (lifecycle_state == LifecycleState::Stopping)
|
||||
{
|
||||
lifecycle_stopped.wait(lock, [this] {
|
||||
return lifecycle_state == LifecycleState::Stopped;
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
lifecycle_state = LifecycleState::Stopping;
|
||||
closed.store(true);
|
||||
}
|
||||
|
||||
stop_started_threads();
|
||||
|
||||
{
|
||||
std::scoped_lock lock(lifecycle_mutex);
|
||||
lifecycle_state = LifecycleState::Stopped;
|
||||
}
|
||||
lifecycle_stopped.notify_all();
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::start_threads(HANDLE token)
|
||||
{
|
||||
inject_thread_start_failure();
|
||||
output_queue_thread = std::thread(&TwoWayPipeMessageIPCImpl::consume_output_queue_thread, this);
|
||||
inject_thread_start_failure();
|
||||
input_queue_thread = std::thread(&TwoWayPipeMessageIPCImpl::consume_input_queue_thread, this);
|
||||
inject_thread_start_failure();
|
||||
input_pipe_thread = std::thread(&TwoWayPipeMessageIPCImpl::start_named_pipe_server, this, token);
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::stop_started_threads()
|
||||
{
|
||||
input_queue.interrupt();
|
||||
if (input_queue_thread.joinable())
|
||||
{
|
||||
input_queue_thread.join();
|
||||
}
|
||||
output_queue.interrupt();
|
||||
cancel_active_output_io();
|
||||
if (output_queue_thread.joinable())
|
||||
{
|
||||
output_queue_thread.join();
|
||||
}
|
||||
{
|
||||
std::scoped_lock lock(pipe_connect_handle_mutex);
|
||||
if (current_connect_pipe_handle != NULL)
|
||||
{
|
||||
// Cancels the pipe currently waiting for a connection.
|
||||
CancelIoEx(current_connect_pipe_handle, NULL);
|
||||
}
|
||||
}
|
||||
if (input_pipe_thread.joinable())
|
||||
{
|
||||
input_pipe_thread.join();
|
||||
}
|
||||
cancel_and_wait_for_connection_handlers();
|
||||
if (pipe_security_token)
|
||||
{
|
||||
CloseHandle(pipe_security_token);
|
||||
pipe_security_token = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::cancel_active_output_io()
|
||||
{
|
||||
std::scoped_lock lock(output_pipe_mutex);
|
||||
if (active_output_pipe_handle != INVALID_HANDLE_VALUE)
|
||||
{
|
||||
CancelIoEx(active_output_pipe_handle, nullptr);
|
||||
}
|
||||
pipe_connect_handle_mutex.unlock();
|
||||
input_pipe_thread.join();
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::send_pipe_message(std::wstring message)
|
||||
{
|
||||
// Adapted from https://learn.microsoft.com/windows/win32/ipc/named-pipe-client
|
||||
HANDLE output_pipe_handle;
|
||||
const wchar_t* message_send = message.c_str();
|
||||
BOOL fSuccess = FALSE;
|
||||
DWORD cbToWrite, cbWritten, dwMode;
|
||||
const wchar_t* lpszPipename = output_pipe_name.c_str();
|
||||
OwnedPipeHandle output_pipe;
|
||||
|
||||
// Try to open a named pipe; wait for it, if necessary.
|
||||
|
||||
while (1)
|
||||
while (!closed.load())
|
||||
{
|
||||
output_pipe_handle = CreateFile(
|
||||
output_pipe.reset(CreateFile(
|
||||
lpszPipename, // pipe name
|
||||
GENERIC_READ | // read and write access
|
||||
GENERIC_WRITE,
|
||||
PipeClientAccess,
|
||||
0, // no sharing
|
||||
NULL, // default security attributes
|
||||
OPEN_EXISTING, // opens existing pipe
|
||||
0, // default attributes
|
||||
NULL); // no template file
|
||||
two_way_pipe_message_ipc::ClientOpenFlags,
|
||||
NULL)); // no template file
|
||||
|
||||
// Break if the pipe handle is valid.
|
||||
|
||||
if (output_pipe_handle != INVALID_HANDLE_VALUE)
|
||||
if (output_pipe.valid())
|
||||
break;
|
||||
|
||||
// Exit if an error other than ERROR_PIPE_BUSY occurs.
|
||||
@@ -111,45 +277,86 @@ void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::send_pipe_message(std::wstr
|
||||
return;
|
||||
}
|
||||
|
||||
// All pipe instances are busy, so wait for 20 seconds.
|
||||
|
||||
if (!WaitNamedPipe(lpszPipename, 20000))
|
||||
// Keep shutdown responsive while the peer pipe has no available instance.
|
||||
#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS
|
||||
if (const HANDLE event = wait_named_pipe_entered_event.load())
|
||||
{
|
||||
SetEvent(event);
|
||||
}
|
||||
#endif
|
||||
if (!WaitNamedPipe(lpszPipename, PipeWaitIntervalMs) && GetLastError() != ERROR_SEM_TIMEOUT)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
dwMode = PIPE_READMODE_MESSAGE;
|
||||
fSuccess = SetNamedPipeHandleState(
|
||||
output_pipe_handle, // pipe handle
|
||||
if (closed.load() || !output_pipe.valid())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
const HANDLE output_pipe_handle = output_pipe.get();
|
||||
const auto clear_active_output_pipe = [&]() {
|
||||
std::scoped_lock lock(output_pipe_mutex);
|
||||
if (active_output_pipe_handle == output_pipe_handle)
|
||||
{
|
||||
active_output_pipe_handle = INVALID_HANDLE_VALUE;
|
||||
}
|
||||
};
|
||||
|
||||
DWORD dwMode = PIPE_READMODE_MESSAGE;
|
||||
if (!SetNamedPipeHandleState(
|
||||
output_pipe_handle,
|
||||
&dwMode, // new pipe mode
|
||||
NULL, // don't set maximum bytes
|
||||
NULL); // don't set maximum time
|
||||
if (!fSuccess)
|
||||
NULL)) // don't set maximum time
|
||||
{
|
||||
clear_active_output_pipe();
|
||||
return;
|
||||
}
|
||||
|
||||
HANDLE write_complete_event = CreateEventW(nullptr, TRUE, FALSE, nullptr);
|
||||
if (!write_complete_event)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
// Send a message to the pipe server.
|
||||
|
||||
cbToWrite = (lstrlen(message_send)) * sizeof(WCHAR); // no need to send final '\0'. Pipe is in message mode.
|
||||
|
||||
fSuccess = WriteFile(
|
||||
output_pipe_handle, // pipe handle
|
||||
message_send, // message
|
||||
cbToWrite, // message length
|
||||
&cbWritten, // bytes written
|
||||
NULL); // not overlapped
|
||||
if (!fSuccess)
|
||||
OVERLAPPED write_overlapped{};
|
||||
write_overlapped.hEvent = write_complete_event;
|
||||
DWORD bytes_written = 0;
|
||||
const DWORD bytes_to_write = (lstrlen(message_send)) * sizeof(WCHAR);
|
||||
BOOL write_succeeded = FALSE;
|
||||
{
|
||||
return;
|
||||
std::scoped_lock lock(output_pipe_mutex);
|
||||
if (closed.load())
|
||||
{
|
||||
CloseHandle(write_complete_event);
|
||||
return;
|
||||
}
|
||||
active_output_pipe_handle = output_pipe_handle;
|
||||
write_succeeded = WriteFile(output_pipe_handle,
|
||||
message_send,
|
||||
bytes_to_write,
|
||||
&bytes_written,
|
||||
&write_overlapped);
|
||||
}
|
||||
CloseHandle(output_pipe_handle);
|
||||
return;
|
||||
if (!write_succeeded)
|
||||
{
|
||||
if (GetLastError() != ERROR_IO_PENDING)
|
||||
{
|
||||
CloseHandle(write_complete_event);
|
||||
clear_active_output_pipe();
|
||||
return;
|
||||
}
|
||||
GetOverlappedResult(output_pipe_handle, &write_overlapped, &bytes_written, TRUE);
|
||||
}
|
||||
|
||||
CloseHandle(write_complete_event);
|
||||
clear_active_output_pipe();
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::consume_output_queue_thread()
|
||||
{
|
||||
while (!closed)
|
||||
while (!closed.load())
|
||||
{
|
||||
std::wstring message = output_queue.pop_message();
|
||||
if (message.length() == 0)
|
||||
@@ -171,6 +378,7 @@ BOOL TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::GetLogonSID(HANDLE hToken,
|
||||
// Verify the parameter passed in is not NULL.
|
||||
if (NULL == ppsid)
|
||||
goto Cleanup;
|
||||
*ppsid = nullptr;
|
||||
|
||||
// Get required buffer size and allocate the TOKEN_GROUPS buffer.
|
||||
|
||||
@@ -222,13 +430,13 @@ BOOL TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::GetLogonSID(HANDLE hToken,
|
||||
if (!CopySid(dwLength, *ppsid, ptg->Groups[dwIndex].Sid))
|
||||
{
|
||||
HeapFree(GetProcessHeap(), 0, static_cast<LPVOID>(*ppsid));
|
||||
*ppsid = nullptr;
|
||||
goto Cleanup;
|
||||
}
|
||||
bSuccess = TRUE;
|
||||
break;
|
||||
}
|
||||
|
||||
bSuccess = TRUE;
|
||||
|
||||
Cleanup:
|
||||
|
||||
// Free the buffer for the token groups.
|
||||
@@ -245,71 +453,128 @@ VOID TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::FreeLogonSID(PSID* ppsid)
|
||||
HeapFree(GetProcessHeap(), 0, static_cast<LPVOID>(*ppsid));
|
||||
}
|
||||
|
||||
int TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::change_pipe_security_allow_restricted_token(HANDLE handle, HANDLE token)
|
||||
bool TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::create_pipe_security_attributes(HANDLE token, PipeSecurityAttributes& security_attributes)
|
||||
{
|
||||
PACL old_dacl, new_dacl;
|
||||
PSECURITY_DESCRIPTOR sd;
|
||||
EXPLICIT_ACCESS ea;
|
||||
PSID user_restricted;
|
||||
int error;
|
||||
HANDLE process_token = nullptr;
|
||||
EXPLICIT_ACCESS entries[3]{};
|
||||
bool success = false;
|
||||
DWORD administrators_sid_size = 0;
|
||||
DWORD local_system_sid_size = 0;
|
||||
TOKEN_ELEVATION elevation{};
|
||||
DWORD elevation_size = 0;
|
||||
PSID server_sid = nullptr;
|
||||
TRUSTEE_TYPE server_trustee_type = TRUSTEE_IS_GROUP;
|
||||
auto set_entry = [](EXPLICIT_ACCESS& entry, DWORD access, PSID sid, TRUSTEE_TYPE trustee_type) {
|
||||
entry.grfAccessPermissions = access;
|
||||
entry.grfAccessMode = SET_ACCESS;
|
||||
entry.grfInheritance = NO_INHERITANCE;
|
||||
entry.Trustee.TrusteeForm = TRUSTEE_IS_SID;
|
||||
entry.Trustee.TrusteeType = trustee_type;
|
||||
entry.Trustee.ptstrName = static_cast<LPTSTR>(sid);
|
||||
};
|
||||
|
||||
if (!GetLogonSID(token, &user_restricted))
|
||||
if (!token)
|
||||
{
|
||||
error = 5; // No access error.
|
||||
goto Ldone;
|
||||
if (!OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &process_token))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
token = process_token;
|
||||
}
|
||||
|
||||
if (GetSecurityInfo(handle,
|
||||
SE_KERNEL_OBJECT,
|
||||
DACL_SECURITY_INFORMATION,
|
||||
NULL,
|
||||
NULL,
|
||||
&old_dacl,
|
||||
NULL,
|
||||
&sd))
|
||||
if (!GetLogonSID(token, &security_attributes.logon_sid))
|
||||
{
|
||||
error = GetLastError();
|
||||
goto Lclean_sid;
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
memset(&ea, 0, sizeof(EXPLICIT_ACCESS));
|
||||
ea.grfAccessPermissions |= GENERIC_READ | FILE_WRITE_ATTRIBUTES;
|
||||
ea.grfAccessPermissions |= GENERIC_WRITE | FILE_READ_ATTRIBUTES;
|
||||
ea.grfAccessPermissions |= SYNCHRONIZE;
|
||||
ea.grfAccessMode = SET_ACCESS;
|
||||
ea.grfInheritance = NO_INHERITANCE;
|
||||
ea.Trustee.TrusteeForm = TRUSTEE_IS_SID;
|
||||
ea.Trustee.TrusteeType = TRUSTEE_IS_USER;
|
||||
ea.Trustee.ptstrName = static_cast<LPTSTR>(user_restricted);
|
||||
|
||||
if (SetEntriesInAcl(1, &ea, old_dacl, &new_dacl))
|
||||
administrators_sid_size = ARRAYSIZE(security_attributes.administrators_sid);
|
||||
local_system_sid_size = ARRAYSIZE(security_attributes.local_system_sid);
|
||||
if (!CreateWellKnownSid(WinBuiltinAdministratorsSid,
|
||||
nullptr,
|
||||
security_attributes.administrators_sid,
|
||||
&administrators_sid_size) ||
|
||||
!CreateWellKnownSid(WinLocalSystemSid,
|
||||
nullptr,
|
||||
security_attributes.local_system_sid,
|
||||
&local_system_sid_size))
|
||||
{
|
||||
error = GetLastError();
|
||||
goto Lclean_sd;
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
if (SetSecurityInfo(handle,
|
||||
SE_KERNEL_OBJECT,
|
||||
DACL_SECURITY_INFORMATION,
|
||||
NULL,
|
||||
NULL,
|
||||
new_dacl,
|
||||
NULL))
|
||||
if (!GetTokenInformation(token, TokenElevation, &elevation, sizeof(elevation), &elevation_size))
|
||||
{
|
||||
error = GetLastError();
|
||||
goto Lclean_dacl;
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
error = 0;
|
||||
server_sid = security_attributes.administrators_sid;
|
||||
if (!elevation.TokenIsElevated)
|
||||
{
|
||||
DWORD token_user_size = 0;
|
||||
GetTokenInformation(token, TokenUser, nullptr, 0, &token_user_size);
|
||||
auto* token_user = static_cast<TOKEN_USER*>(HeapAlloc(GetProcessHeap(), HEAP_ZERO_MEMORY, token_user_size));
|
||||
if (!token_user ||
|
||||
!GetTokenInformation(token, TokenUser, token_user, token_user_size, &token_user_size))
|
||||
{
|
||||
if (token_user)
|
||||
{
|
||||
HeapFree(GetProcessHeap(), 0, token_user);
|
||||
}
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
Lclean_dacl:
|
||||
LocalFree(static_cast<HLOCAL>(new_dacl));
|
||||
Lclean_sd:
|
||||
LocalFree(static_cast<HLOCAL>(sd));
|
||||
Lclean_sid:
|
||||
FreeLogonSID(&user_restricted);
|
||||
Ldone:
|
||||
return error;
|
||||
const DWORD server_sid_size = GetLengthSid(token_user->User.Sid);
|
||||
const bool copied = server_sid_size <= ARRAYSIZE(security_attributes.server_sid) &&
|
||||
CopySid(server_sid_size, security_attributes.server_sid, token_user->User.Sid) == TRUE;
|
||||
HeapFree(GetProcessHeap(), 0, token_user);
|
||||
if (!copied)
|
||||
{
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
server_sid = security_attributes.server_sid;
|
||||
server_trustee_type = TRUSTEE_IS_USER;
|
||||
}
|
||||
|
||||
set_entry(entries[0],
|
||||
FILE_ALL_ACCESS,
|
||||
server_sid,
|
||||
server_trustee_type);
|
||||
set_entry(entries[1],
|
||||
FILE_ALL_ACCESS,
|
||||
security_attributes.local_system_sid,
|
||||
TRUSTEE_IS_USER);
|
||||
set_entry(entries[2],
|
||||
PipeClientAccess,
|
||||
security_attributes.logon_sid,
|
||||
TRUSTEE_IS_USER);
|
||||
|
||||
if (SetEntriesInAcl(ARRAYSIZE(entries), entries, nullptr, &security_attributes.dacl) != ERROR_SUCCESS)
|
||||
{
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
if (!InitializeSecurityDescriptor(&security_attributes.security_descriptor, SECURITY_DESCRIPTOR_REVISION) ||
|
||||
!SetSecurityDescriptorOwner(&security_attributes.security_descriptor, server_sid, FALSE) ||
|
||||
!SetSecurityDescriptorGroup(&security_attributes.security_descriptor, server_sid, FALSE) ||
|
||||
!SetSecurityDescriptorDacl(&security_attributes.security_descriptor,
|
||||
TRUE,
|
||||
security_attributes.dacl,
|
||||
FALSE))
|
||||
{
|
||||
goto Cleanup;
|
||||
}
|
||||
|
||||
security_attributes.attributes.nLength = sizeof(security_attributes.attributes);
|
||||
security_attributes.attributes.lpSecurityDescriptor = &security_attributes.security_descriptor;
|
||||
security_attributes.attributes.bInheritHandle = FALSE;
|
||||
success = true;
|
||||
|
||||
Cleanup:
|
||||
if (process_token)
|
||||
{
|
||||
CloseHandle(process_token);
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
HANDLE TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::create_medium_integrity_token()
|
||||
@@ -348,108 +613,243 @@ HANDLE TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::create_medium_integrity_t
|
||||
return restricted_token_handle;
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::handle_pipe_connection(HANDLE input_pipe_handle)
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::handle_pipe_connection(const std::shared_ptr<ConnectionHandler>& handler)
|
||||
{
|
||||
if (!input_pipe_handle)
|
||||
const HANDLE input_pipe_handle = handler->pipe_handle.get();
|
||||
if (input_pipe_handle == INVALID_HANDLE_VALUE)
|
||||
{
|
||||
finish_connection_handler(handler);
|
||||
return;
|
||||
}
|
||||
constexpr DWORD readBlockBytes = BUFSIZE;
|
||||
std::wstring message;
|
||||
size_t iBlock = 0;
|
||||
message.reserve(BUFSIZE);
|
||||
bool ok;
|
||||
do
|
||||
{
|
||||
constexpr size_t charsPerBlock = readBlockBytes / sizeof(message[0]);
|
||||
message.resize(message.size() + charsPerBlock);
|
||||
DWORD bytesRead = 0;
|
||||
ok = ReadFile(
|
||||
input_pipe_handle,
|
||||
// read the message directly into the string block by block simultaneously resizing it
|
||||
message.data() + iBlock * charsPerBlock,
|
||||
readBlockBytes,
|
||||
&bytesRead,
|
||||
nullptr);
|
||||
|
||||
if (!ok && GetLastError() != ERROR_MORE_DATA)
|
||||
{
|
||||
break;
|
||||
}
|
||||
iBlock++;
|
||||
} while (!ok);
|
||||
// trim the message's buffer
|
||||
const auto nullCharPos = message.find_last_not_of(L'\0');
|
||||
if (nullCharPos != std::wstring::npos)
|
||||
if (!closed.load())
|
||||
{
|
||||
message.resize(nullCharPos + 1);
|
||||
constexpr DWORD readBlockBytes = BUFSIZE;
|
||||
std::wstring message;
|
||||
size_t iBlock = 0;
|
||||
message.reserve(BUFSIZE);
|
||||
bool message_read = false;
|
||||
do
|
||||
{
|
||||
constexpr size_t charsPerBlock = readBlockBytes / sizeof(message[0]);
|
||||
message.resize(message.size() + charsPerBlock);
|
||||
DWORD bytesRead = 0;
|
||||
message_read = ReadFile(
|
||||
input_pipe_handle,
|
||||
message.data() + iBlock * charsPerBlock,
|
||||
readBlockBytes,
|
||||
&bytesRead,
|
||||
nullptr);
|
||||
|
||||
if (!message_read && GetLastError() != ERROR_MORE_DATA)
|
||||
{
|
||||
break;
|
||||
}
|
||||
iBlock++;
|
||||
} while (!message_read);
|
||||
|
||||
if (message_read && !closed.load())
|
||||
{
|
||||
const auto nullCharPos = message.find_last_not_of(L'\0');
|
||||
if (nullCharPos != std::wstring::npos)
|
||||
{
|
||||
message.resize(nullCharPos + 1);
|
||||
}
|
||||
|
||||
input_queue.queue_message(std::move(message));
|
||||
FlushFileBuffers(input_pipe_handle);
|
||||
}
|
||||
}
|
||||
finish_connection_handler(handler);
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::finish_connection_handler(const std::shared_ptr<ConnectionHandler>& handler)
|
||||
{
|
||||
HANDLE pipe_handle = INVALID_HANDLE_VALUE;
|
||||
{
|
||||
std::scoped_lock lock(connection_handlers_mutex);
|
||||
pipe_handle = handler->pipe_handle.get();
|
||||
if (pipe_handle != INVALID_HANDLE_VALUE)
|
||||
{
|
||||
DisconnectNamedPipe(pipe_handle);
|
||||
}
|
||||
handler->pipe_handle.reset();
|
||||
}
|
||||
|
||||
input_queue.queue_message(std::move(message));
|
||||
{
|
||||
std::scoped_lock lock(connection_handlers_mutex);
|
||||
handler->completed = true;
|
||||
}
|
||||
}
|
||||
|
||||
// Flush the pipe to allow the client to read the pipe's contents
|
||||
// before disconnecting. Then disconnect the pipe, and close the
|
||||
// handle to this pipe instance.
|
||||
bool TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::start_connection_handler(OwnedPipeHandle&& pipe_handle)
|
||||
{
|
||||
auto handler = std::make_shared<ConnectionHandler>(std::move(pipe_handle));
|
||||
{
|
||||
std::scoped_lock lock(connection_handlers_mutex);
|
||||
connection_handlers.emplace_back(handler);
|
||||
}
|
||||
|
||||
FlushFileBuffers(input_pipe_handle);
|
||||
DisconnectNamedPipe(input_pipe_handle);
|
||||
CloseHandle(input_pipe_handle);
|
||||
try
|
||||
{
|
||||
handler->thread = std::thread(&TwoWayPipeMessageIPCImpl::handle_pipe_connection, this, handler);
|
||||
return true;
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
finish_connection_handler(handler);
|
||||
reap_finished_connection_handlers();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::reap_finished_connection_handlers()
|
||||
{
|
||||
std::vector<std::shared_ptr<ConnectionHandler>> completed_handlers;
|
||||
{
|
||||
std::scoped_lock lock(connection_handlers_mutex);
|
||||
for (auto it = connection_handlers.begin(); it != connection_handlers.end();)
|
||||
{
|
||||
if ((*it)->completed)
|
||||
{
|
||||
completed_handlers.emplace_back(*it);
|
||||
it = connection_handlers.erase(it);
|
||||
}
|
||||
else
|
||||
{
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (const auto& handler : completed_handlers)
|
||||
{
|
||||
if (handler->thread.joinable())
|
||||
{
|
||||
handler->thread.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::cancel_and_wait_for_connection_handlers()
|
||||
{
|
||||
std::vector<std::shared_ptr<ConnectionHandler>> handlers;
|
||||
{
|
||||
std::scoped_lock lock(connection_handlers_mutex);
|
||||
for (const auto& handler : connection_handlers)
|
||||
{
|
||||
if (handler->pipe_handle.valid())
|
||||
{
|
||||
CancelIoEx(handler->pipe_handle.get(), nullptr);
|
||||
DisconnectNamedPipe(handler->pipe_handle.get());
|
||||
}
|
||||
}
|
||||
handlers.swap(connection_handlers);
|
||||
}
|
||||
|
||||
for (const auto& handler : handlers)
|
||||
{
|
||||
if (handler->thread.joinable())
|
||||
{
|
||||
handler->thread.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::start_named_pipe_server(HANDLE token)
|
||||
{
|
||||
// Adapted from https://learn.microsoft.com/windows/win32/ipc/multithreaded-pipe-server
|
||||
const wchar_t* pipe_name = input_pipe_name.c_str();
|
||||
BOOL connected = FALSE;
|
||||
HANDLE connect_pipe_handle = INVALID_HANDLE_VALUE;
|
||||
while (!closed)
|
||||
// The first instance claims exclusive ownership of the name; later instances omit this flag.
|
||||
auto create_listener = [&](bool first_instance) {
|
||||
DWORD open_mode = PIPE_ACCESS_DUPLEX | WRITE_DAC;
|
||||
if (first_instance)
|
||||
{
|
||||
open_mode |= FILE_FLAG_FIRST_PIPE_INSTANCE;
|
||||
}
|
||||
|
||||
PipeSecurityAttributes security_attributes;
|
||||
if (!create_pipe_security_attributes(token, security_attributes))
|
||||
{
|
||||
return INVALID_HANDLE_VALUE;
|
||||
}
|
||||
|
||||
return CreateNamedPipe(
|
||||
pipe_name,
|
||||
open_mode,
|
||||
PIPE_TYPE_MESSAGE |
|
||||
PIPE_READMODE_MESSAGE |
|
||||
PIPE_WAIT |
|
||||
PIPE_REJECT_REMOTE_CLIENTS,
|
||||
PIPE_UNLIMITED_INSTANCES,
|
||||
BUFSIZE,
|
||||
BUFSIZE,
|
||||
0,
|
||||
&security_attributes.attributes);
|
||||
};
|
||||
|
||||
OwnedPipeHandle listener{ create_listener(true) };
|
||||
if (!listener.valid())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
while (!closed.load())
|
||||
{
|
||||
{
|
||||
std::unique_lock lock(pipe_connect_handle_mutex);
|
||||
connect_pipe_handle = CreateNamedPipe(
|
||||
pipe_name,
|
||||
PIPE_ACCESS_DUPLEX |
|
||||
WRITE_DAC,
|
||||
PIPE_TYPE_MESSAGE |
|
||||
PIPE_READMODE_MESSAGE |
|
||||
PIPE_WAIT,
|
||||
PIPE_UNLIMITED_INSTANCES,
|
||||
BUFSIZE,
|
||||
BUFSIZE,
|
||||
0,
|
||||
NULL);
|
||||
|
||||
if (connect_pipe_handle == INVALID_HANDLE_VALUE)
|
||||
if (closed.load())
|
||||
{
|
||||
return;
|
||||
break;
|
||||
}
|
||||
|
||||
if (token != NULL)
|
||||
{
|
||||
change_pipe_security_allow_restricted_token(connect_pipe_handle, token);
|
||||
}
|
||||
current_connect_pipe_handle = connect_pipe_handle;
|
||||
current_connect_pipe_handle = listener.get();
|
||||
}
|
||||
connected = ConnectNamedPipe(connect_pipe_handle, NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED);
|
||||
const BOOL connected = ConnectNamedPipe(listener.get(), NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED);
|
||||
{
|
||||
std::unique_lock lock(pipe_connect_handle_mutex);
|
||||
current_connect_pipe_handle = NULL;
|
||||
}
|
||||
if (connected)
|
||||
|
||||
if (!connected)
|
||||
{
|
||||
std::thread(&TwoWayPipeMessageIPCImpl::handle_pipe_connection, this, connect_pipe_handle).detach();
|
||||
if (closed.load())
|
||||
{
|
||||
break;
|
||||
}
|
||||
DisconnectNamedPipe(listener.get());
|
||||
continue;
|
||||
}
|
||||
else
|
||||
|
||||
OwnedPipeHandle replacement;
|
||||
while (!closed.load() && !replacement.valid())
|
||||
{
|
||||
// Client could not connect.
|
||||
CloseHandle(connect_pipe_handle);
|
||||
replacement.reset(create_listener(false));
|
||||
if (!replacement.valid() && !closed.load())
|
||||
{
|
||||
Sleep(10);
|
||||
}
|
||||
}
|
||||
|
||||
if (closed.load())
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
start_connection_handler(std::move(listener));
|
||||
listener = std::move(replacement);
|
||||
reap_finished_connection_handlers();
|
||||
}
|
||||
|
||||
if (listener.valid())
|
||||
{
|
||||
DisconnectNamedPipe(listener.get());
|
||||
}
|
||||
reap_finished_connection_handlers();
|
||||
}
|
||||
|
||||
void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::consume_input_queue_thread()
|
||||
{
|
||||
while (!closed)
|
||||
while (!closed.load())
|
||||
{
|
||||
outgoing_message = L"";
|
||||
std::wstring message = input_queue.pop_message();
|
||||
|
||||
Reference in New Issue
Block a user