From ed7595f3a78133e7f635df3fee3c8ca08a24dcfd Mon Sep 17 00:00:00 2001 From: Gordon Lam <73506701+yeelam-gordon@users.noreply.github.com> Date: Sun, 9 Aug 2026 20:56:12 +0800 Subject: [PATCH] Harden IPC pipe ownership and shutdown lifecycle (#48902) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- .../TwoWayPipeMessageIPC.Tests.cpp | 1242 +++++++++++++++++ .../UnitTests-CommonUtils.vcxproj | 6 +- .../UnitTests-CommonUtils.vcxproj.filters | 3 + .../interop/two_way_pipe_message_ipc.cpp | 909 +++++++++--- src/common/interop/two_way_pipe_message_ipc.h | 22 + .../interop/two_way_pipe_message_ipc_impl.h | 124 +- .../two_way_pipe_message_ipc.cpp | 732 +++++++--- 7 files changed, 2694 insertions(+), 344 deletions(-) create mode 100644 src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp diff --git a/src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp b/src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp new file mode 100644 index 0000000000..3f54799ea6 --- /dev/null +++ b/src/common/UnitTests-CommonUtils/TwoWayPipeMessageIPC.Tests.cpp @@ -0,0 +1,1242 @@ +#include "pch.h" + +#include +#include +#include "..\..\modules\Workspaces\WorkspacesLib\IPCHelper.h" + +#include +#include +#include + +using namespace Microsoft::VisualStudio::CppUnitTestFramework; + +namespace UnitTestsCommonUtils +{ + namespace + { + constexpr DWORD PipeClientAccess = FILE_READ_DATA | + FILE_READ_ATTRIBUTES | + READ_CONTROL | + FILE_WRITE_DATA | + FILE_WRITE_ATTRIBUTES | + SYNCHRONIZE; + + std::wstring UniquePipeName() + { + static LONG counter = 0; + return L"\\\\.\\pipe\\pt_ipc_test_" + + std::to_wstring(GetCurrentProcessId()) + L"_" + + std::to_wstring(GetTickCount64()) + L"_" + + std::to_wstring(InterlockedIncrement(&counter)); + } + + std::wstring CurrentExePath() + { + wchar_t path[MAX_PATH * 2]{}; + GetModuleFileNameW(nullptr, path, ARRAYSIZE(path)); + return path; + } + + std::wstring DirectoryOf(const std::wstring& path) + { + const auto separator = path.find_last_of(L"\\/"); + return separator == std::wstring::npos ? path : path.substr(0, separator); + } + + std::wstring BaseNameOf(const std::wstring& path) + { + const auto separator = path.find_last_of(L"\\/"); + return separator == std::wstring::npos ? path : path.substr(separator + 1); + } + + interop_auth::CallerPolicy SelfCallerPolicy() + { + const std::wstring executable = CurrentExePath(); + interop_auth::CallerPolicy policy; + policy.enabled = true; + policy.expectedDirectory = DirectoryOf(executable); + policy.allowedBasenames = { BaseNameOf(executable) }; + policy.requireMicrosoftSignature = false; + return policy; + } + + bool WriteTestMessage(HANDLE pipe) + { + constexpr wchar_t message[] = L"test"; + DWORD bytes_written = 0; + return WriteFile(pipe, + message, + (ARRAYSIZE(message) - 1) * sizeof(wchar_t), + &bytes_written, + nullptr) == TRUE; + } + + void AssertRogueServerCannotImpersonateClient(HANDLE server) + { + if (!ImpersonateNamedPipeClient(server)) + { + return; + } + + HANDLE token = nullptr; + Assert::IsTrue(OpenThreadToken(GetCurrentThread(), TOKEN_QUERY, TRUE, &token) == TRUE, + L"rogue server impersonated the client but could not inspect its token"); + SECURITY_IMPERSONATION_LEVEL level{}; + DWORD level_size = 0; + Assert::IsTrue(GetTokenInformation(token, TokenImpersonationLevel, &level, sizeof(level), &level_size) == TRUE); + CloseHandle(token); + RevertToSelf(); + + Assert::AreEqual(static_cast(SecurityIdentification), static_cast(level), + L"the rogue server received an impersonation-capable client token"); + } + + HANDLE CreateRogueServer(const std::wstring& pipe_name) + { + return CreateNamedPipeW(pipe_name.c_str(), + PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + 1, + 4096, + 4096, + 0, + nullptr); + } + + HANDLE ConnectPipeClient(const std::wstring& pipe_name) + { + constexpr DWORD timeout_ms = 2'000; + const ULONGLONG deadline = GetTickCount64() + timeout_ms; + do + { + HANDLE client = CreateFileW(pipe_name.c_str(), + PipeClientAccess, + 0, + nullptr, + OPEN_EXISTING, + 0, + nullptr); + if (client != INVALID_HANDLE_VALUE) + { + return client; + } + + const DWORD error = GetLastError(); + if (error != ERROR_FILE_NOT_FOUND && error != ERROR_PIPE_BUSY) + { + return INVALID_HANDLE_VALUE; + } + WaitNamedPipeW(pipe_name.c_str(), 50); + } while (GetTickCount64() < deadline); + + SetLastError(ERROR_SEM_TIMEOUT); + return INVALID_HANDLE_VALUE; + } + + struct RestrictedClientToken + { + HANDLE token = nullptr; + + ~RestrictedClientToken() + { + if (token) + { + CloseHandle(token); + } + } + + bool Create() + { + HANDLE process_token = nullptr; + if (!OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY | TOKEN_DUPLICATE, &process_token)) + { + return false; + } + + DWORD user_size = 0; + GetTokenInformation(process_token, TokenUser, nullptr, 0, &user_size); + std::vector user_buffer(user_size); + if (!GetTokenInformation(process_token, TokenUser, user_buffer.data(), user_size, &user_size)) + { + CloseHandle(process_token); + return false; + } + + auto* user = reinterpret_cast(user_buffer.data()); + SID_AND_ATTRIBUTES disabled_sid{ user->User.Sid, 0 }; + HANDLE restricted_primary_token = nullptr; + const BOOL restricted = CreateRestrictedToken(process_token, + 0, + 1, + &disabled_sid, + 0, + nullptr, + 0, + nullptr, + &restricted_primary_token); + CloseHandle(process_token); + if (!restricted) + { + return false; + } + + const BOOL duplicated = DuplicateTokenEx(restricted_primary_token, + TOKEN_QUERY | TOKEN_IMPERSONATE, + nullptr, + SecurityImpersonation, + TokenImpersonation, + &token); + CloseHandle(restricted_primary_token); + return duplicated == TRUE; + } + }; + + struct NormalSameUserClientToken + { + HANDLE token = nullptr; + + ~NormalSameUserClientToken() + { + if (token) + { + CloseHandle(token); + } + } + + bool Create() + { + HANDLE process_token = nullptr; + if (!OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY | TOKEN_DUPLICATE, &process_token)) + { + return false; + } + + BYTE administrators_sid[SECURITY_MAX_SID_SIZE]{}; + DWORD administrators_sid_size = ARRAYSIZE(administrators_sid); + if (!CreateWellKnownSid(WinBuiltinAdministratorsSid, + nullptr, + administrators_sid, + &administrators_sid_size)) + { + CloseHandle(process_token); + return false; + } + + DWORD groups_size = 0; + GetTokenInformation(process_token, TokenGroups, nullptr, 0, &groups_size); + std::vector groups_buffer(groups_size); + if (!GetTokenInformation(process_token, TokenGroups, groups_buffer.data(), groups_size, &groups_size)) + { + CloseHandle(process_token); + return false; + } + + const auto* groups = reinterpret_cast(groups_buffer.data()); + SID_AND_ATTRIBUTES disabled_administrators_sid{}; + DWORD disable_count = 0; + for (DWORD index = 0; index < groups->GroupCount; ++index) + { + if (EqualSid(groups->Groups[index].Sid, administrators_sid)) + { + disabled_administrators_sid.Sid = groups->Groups[index].Sid; + disable_count = 1; + break; + } + } + + HANDLE restricted_primary_token = nullptr; + const BOOL restricted = CreateRestrictedToken(process_token, + 0, + disable_count, + disable_count ? &disabled_administrators_sid : nullptr, + 0, + nullptr, + 0, + nullptr, + &restricted_primary_token); + CloseHandle(process_token); + if (!restricted) + { + return false; + } + + const BOOL duplicated = DuplicateTokenEx(restricted_primary_token, + TOKEN_QUERY | TOKEN_IMPERSONATE, + nullptr, + SecurityImpersonation, + TokenImpersonation, + &token); + CloseHandle(restricted_primary_token); + return duplicated == TRUE; + } + }; + + struct ScopedImpersonation + { + explicit ScopedImpersonation(HANDLE token) : + active(ImpersonateLoggedOnUser(token) == TRUE) + { + } + + ~ScopedImpersonation() + { + if (active) + { + RevertToSelf(); + } + } + + bool active = false; + }; + + std::mutex fault_injection_test_mutex; + + struct FaultInjectionReset + { + std::unique_lock lock{ fault_injection_test_mutex }; + + FaultInjectionReset() + { + two_way_pipe_message_ipc_test::ResetFaultInjection(); + } + + ~FaultInjectionReset() + { + two_way_pipe_message_ipc_test::ResetFaultInjection(); + } + }; + + bool LogonSidPipeAceAllowsInstanceCreation(HANDLE pipe, + HANDLE token, + bool& allows_client_access, + DWORD& matching_access_mask, + DWORD& error) + { + allows_client_access = false; + matching_access_mask = 0; + DWORD groups_size = 0; + GetTokenInformation(token, TokenGroups, nullptr, 0, &groups_size); + std::vector groups_buffer(groups_size); + if (!GetTokenInformation(token, TokenGroups, groups_buffer.data(), groups_size, &groups_size)) + { + error = GetLastError(); + return false; + } + + const auto* groups = reinterpret_cast(groups_buffer.data()); + PSID logon_sid = nullptr; + for (DWORD index = 0; index < groups->GroupCount; ++index) + { + if ((groups->Groups[index].Attributes & SE_GROUP_LOGON_ID) == SE_GROUP_LOGON_ID) + { + logon_sid = groups->Groups[index].Sid; + break; + } + } + if (!logon_sid) + { + error = ERROR_NOT_FOUND; + return false; + } + + PSECURITY_DESCRIPTOR security_descriptor = nullptr; + PACL dacl = nullptr; + const DWORD security_result = GetSecurityInfo(pipe, + SE_KERNEL_OBJECT, + DACL_SECURITY_INFORMATION, + nullptr, + nullptr, + &dacl, + nullptr, + &security_descriptor); + if (security_result != ERROR_SUCCESS) + { + error = security_result; + return false; + } + + bool allows_creation = false; + ACL_SIZE_INFORMATION acl_info{}; + if (!GetAclInformation(dacl, &acl_info, sizeof(acl_info), AclSizeInformation)) + { + error = GetLastError(); + LocalFree(security_descriptor); + return false; + } + + for (DWORD index = 0; index < acl_info.AceCount; ++index) + { + void* ace = nullptr; + if (!GetAce(dacl, index, &ace)) + { + error = GetLastError(); + LocalFree(security_descriptor); + return false; + } + + auto* allowed_ace = static_cast(ace); + if (allowed_ace->Header.AceType != ACCESS_ALLOWED_ACE_TYPE || + !EqualSid(logon_sid, reinterpret_cast(&allowed_ace->SidStart))) + { + continue; + } + + const DWORD access_mask = allowed_ace->Mask; + matching_access_mask |= access_mask; + allows_creation |= (access_mask & (GENERIC_WRITE | FILE_CREATE_PIPE_INSTANCE)) != 0; + allows_client_access |= (access_mask & PipeClientAccess) == PipeClientAccess; + } + LocalFree(security_descriptor); + error = ERROR_SUCCESS; + return allows_creation; + } + + struct OccupiedPipe + { + std::wstring name = UniquePipeName(); + HANDLE server = INVALID_HANDLE_VALUE; + HANDLE client = INVALID_HANDLE_VALUE; + + ~OccupiedPipe() + { + if (client != INVALID_HANDLE_VALUE) + { + CloseHandle(client); + } + if (server != INVALID_HANDLE_VALUE) + { + CloseHandle(server); + } + } + + bool Create() + { + server = CreateNamedPipeW(name.c_str(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + PIPE_UNLIMITED_INSTANCES, + 4096, + 4096, + 0, + nullptr); + if (server == INVALID_HANDLE_VALUE) + { + return false; + } + + std::thread connectThread([&]() { + client = CreateFileW(name.c_str(), GENERIC_READ | GENERIC_WRITE, 0, nullptr, OPEN_EXISTING, 0, nullptr); + }); + const BOOL connected = ConnectNamedPipe(server, nullptr) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED); + connectThread.join(); + return connected && client != INVALID_HANDLE_VALUE; + } + }; + + struct NonReadingPipePeer + { + std::wstring name = UniquePipeName(); + HANDLE server = INVALID_HANDLE_VALUE; + HANDLE connected = CreateEventW(nullptr, TRUE, FALSE, nullptr); + std::thread accept_thread; + + ~NonReadingPipePeer() + { + if (server != INVALID_HANDLE_VALUE) + { + DisconnectNamedPipe(server); + CloseHandle(server); + } + if (accept_thread.joinable()) + { + accept_thread.join(); + } + if (connected) + { + CloseHandle(connected); + } + } + + bool Start() + { + server = CreateNamedPipeW(name.c_str(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + 1, + 1, + 1, + 0, + nullptr); + if (server == INVALID_HANDLE_VALUE) + { + return false; + } + + accept_thread = std::thread([this]() { + const BOOL accepted = ConnectNamedPipe(server, nullptr) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED); + if (accepted) + { + SetEvent(connected); + } + }); + return true; + } + }; + + struct BlockedRejectedConnection + { + HANDLE client = INVALID_HANDLE_VALUE; + HANDLE handler_entered = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE allow_handler_to_finish = CreateEventW(nullptr, TRUE, FALSE, nullptr); + + ~BlockedRejectedConnection() + { + if (client != INVALID_HANDLE_VALUE) + { + CloseHandle(client); + } + if (handler_entered) + { + CloseHandle(handler_entered); + } + if (allow_handler_to_finish) + { + CloseHandle(allow_handler_to_finish); + } + } + + bool Start(TwoWayPipeMessageIPC& server, const std::wstring& input_pipe_name) + { + interop_auth::CallerPolicy policy; + policy.enabled = true; + policy.expectedDirectory = L"Z:\\not-the-test-host"; + policy.allowedBasenames = { L"not-the-test-host.exe" }; + policy.requireMicrosoftSignature = false; + policy.logReject = [this](const interop_auth::AuthResult&) { + SetEvent(handler_entered); + WaitForSingleObject(allow_handler_to_finish, 10'000); + }; + + server.start(nullptr, policy); + client = ConnectPipeClient(input_pipe_name); + return client != INVALID_HANDLE_VALUE && + WaitForSingleObject(handler_entered, 2'000) == WAIT_OBJECT_0; + } + + void AllowHandlerToFinish() + { + SetEvent(allow_handler_to_finish); + } + }; + } + + TEST_CLASS(TwoWayPipeMessageIPCTests) + { + public: + TEST_METHOD(ServerDoesNotJoinAnExistingPipeName) + { + OccupiedPipe occupiedPipe; + Assert::IsTrue(occupiedPipe.Create(), L"failed to occupy the pipe name"); + + TwoWayPipeMessageIPC server(occupiedPipe.name, UniquePipeName(), nullptr); + server.start(nullptr); + + // The existing instance is busy. A server that wrongly creates a second instance makes + // WaitNamedPipe succeed; FILE_FLAG_FIRST_PIPE_INSTANCE must instead make its first + // CreateNamedPipe call fail and leave no available instance. + const BOOL available = WaitNamedPipeW(occupiedPipe.name.c_str(), 2000); + server.end(); + + Assert::IsFalse(available, L"the server must not join an existing pipe name"); + } + + TEST_METHOD(CommonOutboundPipeClientUsesIdentificationQos) + { + const std::wstring rogue_pipe_name = UniquePipeName(); + HANDLE rogue_server = CreateRogueServer(rogue_pipe_name); + Assert::IsTrue(rogue_server != INVALID_HANDLE_VALUE, L"failed to create the rogue common IPC server"); + + TwoWayPipeMessageIPC client(UniquePipeName(), rogue_pipe_name, nullptr); + client.start(nullptr); + client.send(L"message"); + + const BOOL connected = ConnectNamedPipe(rogue_server, nullptr) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED); + Assert::IsTrue(connected == TRUE, L"the common IPC client did not connect to the rogue server"); + AssertRogueServerCannotImpersonateClient(rogue_server); + + wchar_t message[16]{}; + DWORD bytes_read = 0; + ReadFile(rogue_server, message, sizeof(message), &bytes_read, nullptr); + client.end(); + DisconnectNamedPipe(rogue_server); + CloseHandle(rogue_server); + } + + TEST_METHOD(WorkspacesLauncherArrangerClientUsesIdentificationQos) + { + const std::wstring& pipe_name = IPCHelperStrings::LauncherArrangerPipeName; + HANDLE rogue_server = CreateRogueServer(pipe_name); + Assert::IsTrue(rogue_server != INVALID_HANDLE_VALUE, + L"failed to claim the static LauncherArranger pipe name for the rogue server"); + + HANDLE client = INVALID_HANDLE_VALUE; + std::thread connect_thread([&]() { + client = CreateFileW(pipe_name.c_str(), + PipeClientAccess, + 0, + nullptr, + OPEN_EXISTING, + two_way_pipe_message_ipc::ClientOpenFlags, + nullptr); + }); + const BOOL connected = ConnectNamedPipe(rogue_server, nullptr) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED); + connect_thread.join(); + + Assert::IsTrue(connected == TRUE && client != INVALID_HANDLE_VALUE, + L"the LauncherArranger client could not connect to the rogue server"); + AssertRogueServerCannotImpersonateClient(rogue_server); + + CloseHandle(client); + DisconnectNamedPipe(rogue_server); + CloseHandle(rogue_server); + } + + TEST_METHOD(RestrictedClientCanConnectButCannotCreateAnotherServerInstance) + { + HANDLE token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &token) == TRUE, + L"failed to open the current process token"); + + RestrictedClientToken restricted_client; + Assert::IsTrue(restricted_client.Create(), L"failed to create the restricted same-logon client token"); + + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + server.start(token); + + { + ScopedImpersonation impersonation(restricted_client.token); + Assert::IsTrue(impersonation.active, L"failed to impersonate the restricted client token"); + + HANDLE client = ConnectPipeClient(input_pipe_name); + const DWORD connect_error = GetLastError(); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, + (L"the explicitly-permitted client access must connect; error=" + + std::to_wstring(connect_error)) + .c_str()); + + // A later CreateNamedPipe call is authorized by the first instance's DACL. Verify + // that the ACE for this same-logon client contains every requested client right + // but excludes FILE_CREATE_PIPE_INSTANCE (also included by GENERIC_WRITE). + DWORD acl_error = ERROR_SUCCESS; + bool acl_allows_client_access = false; + DWORD matching_access_mask = 0; + bool can_create_later_instance = false; + const ULONGLONG acl_deadline = GetTickCount64() + 2'000; + do + { + can_create_later_instance = LogonSidPipeAceAllowsInstanceCreation(client, + token, + acl_allows_client_access, + matching_access_mask, + acl_error); + if (acl_error != ERROR_SUCCESS || acl_allows_client_access) + { + break; + } + Sleep(10); + } while (GetTickCount64() < acl_deadline); + CloseHandle(client); + + Assert::IsTrue(acl_allows_client_access, + (L"the same-logon client ACE must contain the explicit client access rights; mask=" + + std::to_wstring(matching_access_mask)) + .c_str()); + Assert::IsFalse(can_create_later_instance, + L"a same-logon client must not create a later pipe instance"); + Assert::AreEqual(static_cast(ERROR_SUCCESS), acl_error); + } + + server.end(); + CloseHandle(token); + } + + TEST_METHOD(NormalSameUserCannotModifyProtectedDaclOrCreateAnotherServerInstance) + { + HANDLE token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &token) == TRUE); + + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + server.start(token); + CloseHandle(token); + + NormalSameUserClientToken normal_client; + Assert::IsTrue(normal_client.Create(), L"failed to create the normal same-user client token"); + + { + ScopedImpersonation impersonation(normal_client.token); + Assert::IsTrue(impersonation.active, L"failed to impersonate the normal same-user client token"); + + HANDLE client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, L"the normal client could not connect to the protected pipe"); + + PSECURITY_DESCRIPTOR security_descriptor = nullptr; + PSID owner = nullptr; + PACL dacl = nullptr; + Assert::AreEqual(static_cast(ERROR_SUCCESS), + GetSecurityInfo(client, + SE_KERNEL_OBJECT, + OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION, + &owner, + nullptr, + &dacl, + nullptr, + &security_descriptor)); + + BYTE administrators_sid[SECURITY_MAX_SID_SIZE]{}; + DWORD administrators_sid_size = ARRAYSIZE(administrators_sid); + Assert::IsTrue(CreateWellKnownSid(WinBuiltinAdministratorsSid, + nullptr, + administrators_sid, + &administrators_sid_size) == TRUE); + Assert::IsTrue(EqualSid(owner, administrators_sid) == TRUE, + L"the pipe owner must not be the normal client user"); + + const DWORD set_dacl_error = SetSecurityInfo(client, + SE_KERNEL_OBJECT, + DACL_SECURITY_INFORMATION, + nullptr, + nullptr, + dacl, + nullptr); + SetLastError(ERROR_SUCCESS); + HANDLE rogue_server = CreateNamedPipeW(input_pipe_name.c_str(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + PIPE_UNLIMITED_INSTANCES, + 4096, + 4096, + 0, + nullptr); + const DWORD create_instance_error = GetLastError(); + if (rogue_server != INVALID_HANDLE_VALUE) + { + CloseHandle(rogue_server); + } + + LocalFree(security_descriptor); + CloseHandle(client); + + Assert::AreEqual(static_cast(ERROR_ACCESS_DENIED), set_dacl_error); + Assert::IsTrue(rogue_server == INVALID_HANDLE_VALUE, + L"the normal same-user client created a later server instance"); + Assert::AreEqual(static_cast(ERROR_ACCESS_DENIED), create_instance_error); + } + + server.end(); + } + + TEST_METHOD(RejectedClientRapidCloseNeverReleasesPipeName) + { + HANDLE token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &token) == TRUE); + + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + interop_auth::CallerPolicy reject_policy; + reject_policy.enabled = true; + reject_policy.expectedDirectory = L"Z:\\not-the-test-host"; + reject_policy.allowedBasenames = { L"not-the-test-host.exe" }; + reject_policy.requireMicrosoftSignature = false; + server.start(token, reject_policy); + CloseHandle(token); + + NormalSameUserClientToken normal_client; + Assert::IsTrue(normal_client.Create(), L"failed to create the normal same-user client token"); + { + ScopedImpersonation impersonation(normal_client.token); + Assert::IsTrue(impersonation.active, L"failed to impersonate the normal same-user client token"); + + HANDLE client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, L"the rejected client could not connect"); + CloseHandle(client); + + for (int attempt = 0; attempt < 100; ++attempt) + { + SetLastError(ERROR_SUCCESS); + HANDLE rogue_server = CreateNamedPipeW(input_pipe_name.c_str(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + PIPE_UNLIMITED_INSTANCES, + 4096, + 4096, + 0, + nullptr); + const DWORD create_error = GetLastError(); + if (rogue_server != INVALID_HANDLE_VALUE) + { + CloseHandle(rogue_server); + } + + Assert::IsTrue(rogue_server == INVALID_HANDLE_VALUE, + L"the pipe name was released while a rejected client closed"); + Assert::AreEqual(static_cast(ERROR_ACCESS_DENIED), create_error); + Sleep(1); + } + } + + server.end(); + } + + TEST_METHOD(ReplacementListenerIsReservedBeforeRejectedHandlerStarts) + { + FaultInjectionReset reset; + HANDLE before_replacement = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE allow_replacement = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE handler_rejected = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(before_replacement); + Assert::IsNotNull(allow_replacement); + Assert::IsNotNull(handler_rejected); + two_way_pipe_message_ipc_test::SetBeforeReplacementListenerEvents(before_replacement, allow_replacement); + + HANDLE token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &token) == TRUE); + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + interop_auth::CallerPolicy reject_policy; + reject_policy.enabled = true; + reject_policy.expectedDirectory = L"Z:\\not-the-test-host"; + reject_policy.allowedBasenames = { L"not-the-test-host.exe" }; + reject_policy.requireMicrosoftSignature = false; + reject_policy.logReject = [handler_rejected](const interop_auth::AuthResult&) { + SetEvent(handler_rejected); + }; + server.start(token, reject_policy); + CloseHandle(token); + + NormalSameUserClientToken normal_client; + Assert::IsTrue(normal_client.Create(), L"failed to create the normal same-user client token"); + { + ScopedImpersonation impersonation(normal_client.token); + Assert::IsTrue(impersonation.active, L"failed to impersonate the normal same-user client token"); + HANDLE client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, L"the rejected client could not connect"); + + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(before_replacement, 2'000), + L"the server did not begin reserving a replacement listener"); + Assert::AreEqual(static_cast(WAIT_TIMEOUT), WaitForSingleObject(handler_rejected, 0), + L"the rejected handler started before its replacement listener was reserved"); + + SetLastError(ERROR_SUCCESS); + HANDLE rogue_server = CreateNamedPipeW(input_pipe_name.c_str(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + PIPE_UNLIMITED_INSTANCES, + 4096, + 4096, + 0, + nullptr); + const DWORD create_error = GetLastError(); + if (rogue_server != INVALID_HANDLE_VALUE) + { + CloseHandle(rogue_server); + } + Assert::IsTrue(rogue_server == INVALID_HANDLE_VALUE, + L"the pipe name was released before the replacement listener existed"); + Assert::AreEqual(static_cast(ERROR_ACCESS_DENIED), create_error); + CloseHandle(client); + } + + SetEvent(allow_replacement); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(handler_rejected, 2'000), + L"the rejected handler did not run after the replacement was created"); + server.end(); + two_way_pipe_message_ipc_test::SetBeforeReplacementListenerEvents(nullptr, nullptr); + CloseHandle(before_replacement); + CloseHandle(allow_replacement); + CloseHandle(handler_rejected); + } + + TEST_METHOD(OwnedSecurityTokenSupportsReplacementAfterCallerClosesIt) + { + HANDLE caller_token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &caller_token) == TRUE); + + const std::wstring input_pipe_name = UniquePipeName(); + HANDLE first_client_dispatched = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE two_clients_dispatched = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(first_client_dispatched); + Assert::IsNotNull(two_clients_dispatched); + std::atomic dispatch_count = 0; + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), [&](const std::wstring&) { + const int count = ++dispatch_count; + if (count == 1) + { + SetEvent(first_client_dispatched); + } + else if (count == 2) + { + SetEvent(two_clients_dispatched); + } + }); + server.start(caller_token, SelfCallerPolicy()); + CloseHandle(caller_token); + + HANDLE first_client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(first_client != INVALID_HANDLE_VALUE, L"the first client could not connect"); + Assert::IsTrue(WriteTestMessage(first_client), L"the first client could not write"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(first_client_dispatched, 2'000), + L"the first client was not authenticated and dispatched"); + CloseHandle(first_client); + HANDLE second_client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(second_client != INVALID_HANDLE_VALUE, + L"the replacement listener did not survive the caller token closing"); + Assert::IsTrue(WriteTestMessage(second_client), L"the second client could not write"); + + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(two_clients_dispatched, 2'000), + L"the replacement listener did not authenticate the second client"); + CloseHandle(second_client); + server.end(); + CloseHandle(first_client_dispatched); + CloseHandle(two_clients_dispatched); + } + + TEST_METHOD(ShutdownClosesReplacementReservedDuringHandoff) + { + FaultInjectionReset reset; + HANDLE after_replacement = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE allow_handoff = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE shutdown_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(after_replacement); + Assert::IsNotNull(allow_handoff); + Assert::IsNotNull(shutdown_finished); + two_way_pipe_message_ipc_test::SetAfterReplacementListenerEvents(after_replacement, allow_handoff); + + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + server.start(nullptr); + HANDLE client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, L"the handoff client could not connect"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(after_replacement, 2'000), + L"the replacement listener was not created"); + + std::thread shutdown_thread([&]() { + server.end(); + SetEvent(shutdown_finished); + }); + Assert::AreEqual(static_cast(WAIT_TIMEOUT), WaitForSingleObject(shutdown_finished, 200), + L"shutdown unexpectedly completed before the handoff race was released"); + SetEvent(allow_handoff); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(shutdown_finished, 5'000), + L"shutdown did not close the reserved replacement listener"); + shutdown_thread.join(); + CloseHandle(client); + + HANDLE probe = CreateNamedPipeW(input_pipe_name.c_str(), + PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, + 1, + 4096, + 4096, + 0, + nullptr); + Assert::IsTrue(probe != INVALID_HANDLE_VALUE, + L"shutdown leaked a replacement listener reservation"); + CloseHandle(probe); + + two_way_pipe_message_ipc_test::SetAfterReplacementListenerEvents(nullptr, nullptr); + CloseHandle(after_replacement); + CloseHandle(allow_handoff); + CloseHandle(shutdown_finished); + } + + TEST_METHOD(HandlerThreadStartFailureTransfersAndClosesPipeOnce) + { + FaultInjectionReset reset; + HANDLE handler_start_attempted = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(handler_start_attempted); + two_way_pipe_message_ipc_test::FailHandlerThreadStartAfter(0); + two_way_pipe_message_ipc_test::SetHandlerThreadStartAttemptEvent(handler_start_attempted); + + const std::wstring input_pipe_name = UniquePipeName(); + HANDLE dispatched = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(dispatched); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), [dispatched](const std::wstring&) { + SetEvent(dispatched); + }); + server.start(nullptr); + + HANDLE first_client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(first_client != INVALID_HANDLE_VALUE, L"the first client could not connect"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(handler_start_attempted, 2'000), + L"the injected handler-start failure was not consumed for the first client"); + CloseHandle(first_client); + HANDLE second_client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(second_client != INVALID_HANDLE_VALUE, + L"the listener did not remain usable after handler thread creation failed"); + Assert::IsTrue(WriteTestMessage(second_client), L"the second client could not write"); + + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(dispatched, 2'000), + L"the replacement listener did not process the second client"); + CloseHandle(second_client); + server.end(); + two_way_pipe_message_ipc_test::SetHandlerThreadStartAttemptEvent(nullptr); + CloseHandle(handler_start_attempted); + CloseHandle(dispatched); + } + + TEST_METHOD(StartFailureAfterFirstThreadCleansUp) + { + FaultInjectionReset reset; + auto server = std::make_unique(UniquePipeName(), UniquePipeName(), nullptr); + two_way_pipe_message_ipc_test::FailThreadStartAfter(1); + + bool threw = false; + try + { + server->start(nullptr); + } + catch (const std::system_error&) + { + threw = true; + } + + Assert::IsTrue(threw, L"the injected second thread creation failure was not observed"); + server->end(); + server.reset(); + } + + TEST_METHOD(StartFailureAfterSecondThreadCleansUp) + { + FaultInjectionReset reset; + auto server = std::make_unique(UniquePipeName(), UniquePipeName(), nullptr); + two_way_pipe_message_ipc_test::FailThreadStartAfter(2); + + bool threw = false; + try + { + server->start(nullptr); + } + catch (const std::system_error&) + { + threw = true; + } + + Assert::IsTrue(threw, L"the injected third thread creation failure was not observed"); + server->end(); + server.reset(); + } + + TEST_METHOD(EndWaitsForActiveConnectionHandler) + { + const std::wstring input_pipe_name = UniquePipeName(); + TwoWayPipeMessageIPC server(input_pipe_name, UniquePipeName(), nullptr); + BlockedRejectedConnection connection; + Assert::IsTrue(connection.Start(server, input_pipe_name), + L"the test connection did not enter its handler"); + + HANDLE end_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(end_finished); + std::thread shutdown_thread([&]() { + server.end(); + SetEvent(end_finished); + }); + + Assert::AreEqual(static_cast(WAIT_TIMEOUT), WaitForSingleObject(end_finished, 200), + L"end must wait for the active handler before returning"); + connection.AllowHandlerToFinish(); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(end_finished, 5'000), + L"end did not finish after the active handler completed"); + + shutdown_thread.join(); + CloseHandle(end_finished); + } + + TEST_METHOD(DestructorWaitsForActiveConnectionHandler) + { + const std::wstring input_pipe_name = UniquePipeName(); + auto server = std::make_unique(input_pipe_name, UniquePipeName(), nullptr); + BlockedRejectedConnection connection; + Assert::IsTrue(connection.Start(*server, input_pipe_name), + L"the test connection did not enter its handler"); + + HANDLE destructor_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(destructor_finished); + std::thread destroyer([&]() { + server.reset(); + SetEvent(destructor_finished); + }); + + Assert::AreEqual(static_cast(WAIT_TIMEOUT), WaitForSingleObject(destructor_finished, 200), + L"destruction must wait for the active handler before freeing IPC state"); + connection.AllowHandlerToFinish(); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(destructor_finished, 5'000), + L"destruction did not finish after the active handler completed"); + + destroyer.join(); + CloseHandle(destructor_finished); + } + + TEST_METHOD(DestructorJoinsHandlerAfterCompletion) + { + FaultInjectionReset reset; + HANDLE handler_completed = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE allow_handler_return = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE destructor_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE handler_rejected = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(handler_completed); + Assert::IsNotNull(allow_handler_return); + Assert::IsNotNull(destructor_finished); + Assert::IsNotNull(handler_rejected); + two_way_pipe_message_ipc_test::SetHandlerCompletionEvents(handler_completed, allow_handler_return); + + const std::wstring input_pipe_name = UniquePipeName(); + auto server = std::make_unique(input_pipe_name, UniquePipeName(), nullptr); + interop_auth::CallerPolicy reject_policy; + reject_policy.enabled = true; + reject_policy.expectedDirectory = L"Z:\\not-the-test-host"; + reject_policy.allowedBasenames = { L"not-the-test-host.exe" }; + reject_policy.requireMicrosoftSignature = false; + reject_policy.logReject = [handler_rejected](const interop_auth::AuthResult&) { + SetEvent(handler_rejected); + }; + server->start(nullptr, reject_policy); + + HANDLE client = ConnectPipeClient(input_pipe_name); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, L"the rejected client could not connect"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(handler_rejected, 2'000), + L"the handler did not reject the test client"); + CloseHandle(client); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(handler_completed, 2'000), + L"the handler did not reach its completion point"); + + std::thread destroyer([&]() { + server.reset(); + SetEvent(destructor_finished); + }); + + Assert::AreEqual(static_cast(WAIT_TIMEOUT), WaitForSingleObject(destructor_finished, 200), + L"destruction returned before the completed handler thread was joined"); + SetEvent(allow_handler_return); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(destructor_finished, 5'000)); + + destroyer.join(); + two_way_pipe_message_ipc_test::SetHandlerCompletionEvents(nullptr, nullptr); + CloseHandle(handler_completed); + CloseHandle(allow_handler_return); + CloseHandle(destructor_finished); + CloseHandle(handler_rejected); + } + + TEST_METHOD(DestructorCancelsBlockedConnectionRead) + { + const std::wstring input_pipe_name = UniquePipeName(); + auto server = std::make_unique(input_pipe_name, UniquePipeName(), nullptr); + HANDLE server_token = nullptr; + Assert::IsTrue(OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &server_token) == TRUE); + server->start(server_token); + CloseHandle(server_token); + + RestrictedClientToken restricted_client; + Assert::IsTrue(restricted_client.Create(), L"failed to create the restricted same-logon client token"); + + HANDLE client = INVALID_HANDLE_VALUE; + { + ScopedImpersonation impersonation(restricted_client.token); + Assert::IsTrue(impersonation.active, L"failed to impersonate the restricted client token"); + client = ConnectPipeClient(input_pipe_name); + } + const DWORD connect_error = GetLastError(); + Assert::IsTrue(client != INVALID_HANDLE_VALUE, + (L"failed to connect the client that blocks in ReadFile; error=" + + std::to_wstring(connect_error)) + .c_str()); + + // The next listener is created only after the accepted connection has been registered + // for lifetime tracking, so destruction must cancel that handler's blocked read. + Assert::IsTrue(WaitNamedPipeW(input_pipe_name.c_str(), 2'000) == TRUE, + L"the server did not create the next listening instance"); + + HANDLE destructor_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(destructor_finished); + std::thread destroyer([&]() { + server.reset(); + SetEvent(destructor_finished); + }); + + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(destructor_finished, 5'000), + L"destruction did not cancel and join the handler blocked in ReadFile"); + + destroyer.join(); + CloseHandle(destructor_finished); + CloseHandle(client); + } + + TEST_METHOD(EndInterruptsBusyOutputPipeWait) + { + FaultInjectionReset reset; + OccupiedPipe busy_output_pipe; + Assert::IsTrue(busy_output_pipe.Create(), L"failed to create the busy output pipe"); + + HANDLE wait_entered = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(wait_entered); + two_way_pipe_message_ipc_test::SetWaitNamedPipeEnteredEvent(wait_entered); + + TwoWayPipeMessageIPC server(UniquePipeName(), busy_output_pipe.name, nullptr); + server.start(nullptr); + server.send(L"message"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(wait_entered, 2'000), + L"the output worker did not enter WaitNamedPipe"); + + const auto start = std::chrono::steady_clock::now(); + server.end(); + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start); + + two_way_pipe_message_ipc_test::SetWaitNamedPipeEnteredEvent(nullptr); + CloseHandle(wait_entered); + Assert::IsTrue(elapsed.count() < 1'000, + L"end waited too long for an unavailable output pipe"); + } + + TEST_METHOD(DestructorCancelsPendingOutputWrite) + { + FaultInjectionReset reset; + NonReadingPipePeer peer; + Assert::IsTrue(peer.Start(), L"failed to create the non-reading output peer"); + + HANDLE write_pending = CreateEventW(nullptr, TRUE, FALSE, nullptr); + HANDLE destructor_finished = CreateEventW(nullptr, TRUE, FALSE, nullptr); + Assert::IsNotNull(write_pending); + Assert::IsNotNull(destructor_finished); + two_way_pipe_message_ipc_test::SetOutputWritePendingEvent(write_pending); + + auto server = std::make_unique(UniquePipeName(), peer.name, nullptr); + server->start(nullptr); + server->send(std::wstring(512 * 1024, L'x')); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(write_pending, 5'000), + L"the output write did not become pending against the non-reading peer"); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(peer.connected, 5'000), + L"the output peer did not accept the connection"); + peer.accept_thread.join(); + + const auto start = std::chrono::steady_clock::now(); + std::thread destroyer([&]() { + server.reset(); + SetEvent(destructor_finished); + }); + Assert::AreEqual(static_cast(WAIT_OBJECT_0), WaitForSingleObject(destructor_finished, 1'000), + L"destruction did not cancel the pending output write"); + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start); + + destroyer.join(); + two_way_pipe_message_ipc_test::SetOutputWritePendingEvent(nullptr); + CloseHandle(write_pending); + CloseHandle(destructor_finished); + Assert::IsTrue(elapsed.count() < 1'000, + L"destruction waited too long for the non-reading output peer"); + } + }; +} diff --git a/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj b/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj index 1fd90f31ae..44c9ddf3fc 100644 --- a/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj +++ b/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj @@ -28,7 +28,7 @@ ..\;..\utils;..\Telemetry;..\..\;..\..\..\deps\;..\..\..\packages\Microsoft.Windows.ImplementationLibrary.1.0.260126.7\include;$(VCInstallDir)UnitTest\include;%(AdditionalIncludeDirectories) stdcpp23 - SPDLOG_WCHAR_TO_UTF8_SUPPORT;SPDLOG_HEADER_ONLY;%(PreprocessorDefinitions) + SPDLOG_WCHAR_TO_UTF8_SUPPORT;SPDLOG_HEADER_ONLY;TWO_WAY_PIPE_MESSAGE_IPC_TESTS;%(PreprocessorDefinitions) $(VCInstallDir)UnitTest\lib;%(AdditionalLibraryDirectories) @@ -73,6 +73,10 @@ + + + NotUsing + diff --git a/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj.filters b/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj.filters index c642faa4b5..c34c46a1a7 100644 --- a/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj.filters +++ b/src/common/UnitTests-CommonUtils/UnitTests-CommonUtils.vcxproj.filters @@ -120,6 +120,9 @@ Source Files\Integration + + Source Files\Integration + diff --git a/src/common/interop/two_way_pipe_message_ipc.cpp b/src/common/interop/two_way_pipe_message_ipc.cpp index 1a716c9d6f..464ac680f9 100644 --- a/src/common/interop/two_way_pipe_message_ipc.cpp +++ b/src/common/interop/two_way_pipe_message_ipc.cpp @@ -1,9 +1,171 @@ #include "pch.h" #include "two_way_pipe_message_ipc_impl.h" +#include #include +#include 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 wait_named_pipe_entered_event{ nullptr }; + std::atomic handler_completed_event{ nullptr }; + std::atomic handler_allow_return_event{ nullptr }; + std::atomic before_replacement_listener_event{ nullptr }; + std::atomic allow_replacement_listener_event{ nullptr }; + std::atomic after_replacement_listener_event{ nullptr }; + std::atomic allow_after_replacement_listener_event{ nullptr }; + std::atomic_int handler_thread_start_failure_after{ -1 }; + std::atomic handler_thread_start_attempt_event{ nullptr }; + std::atomic output_write_pending_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; + } + } + } + + void inject_handler_thread_start_failure() + { + int remaining = handler_thread_start_failure_after.load(); + while (remaining >= 0) + { + if (remaining == 0) + { + if (handler_thread_start_failure_after.compare_exchange_weak(remaining, -1)) + { + if (const HANDLE attempt_event = handler_thread_start_attempt_event.load()) + { + SetEvent(attempt_event); + } + throw std::system_error(std::make_error_code(std::errc::resource_unavailable_try_again)); + } + continue; + } + if (handler_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 SetHandlerCompletionEvents(HANDLE completed_event, HANDLE allow_return_event) + { + handler_completed_event.store(completed_event); + handler_allow_return_event.store(allow_return_event); + } + + void SetBeforeReplacementListenerEvents(HANDLE reached_event, HANDLE allow_creation_event) + { + before_replacement_listener_event.store(reached_event); + allow_replacement_listener_event.store(allow_creation_event); + } + + void SetAfterReplacementListenerEvents(HANDLE reached_event, HANDLE allow_continue_event) + { + after_replacement_listener_event.store(reached_event); + allow_after_replacement_listener_event.store(allow_continue_event); + } + + void FailHandlerThreadStartAfter(int successful_starts) + { + handler_thread_start_failure_after.store(successful_starts); + } + + void SetHandlerThreadStartAttemptEvent(HANDLE event) + { + handler_thread_start_attempt_event.store(event); + } + + void SetOutputWritePendingEvent(HANDLE event) + { + output_write_pending_event.store(event); + } + + void ResetFaultInjection() + { + thread_start_failure_after.store(-1); + wait_named_pipe_entered_event.store(nullptr); + handler_completed_event.store(nullptr); + handler_allow_return_event.store(nullptr); + before_replacement_listener_event.store(nullptr); + allow_replacement_listener_event.store(nullptr); + after_replacement_listener_event.store(nullptr); + allow_after_replacement_listener_event.store(nullptr); + handler_thread_start_failure_after.store(-1); + handler_thread_start_attempt_event.store(nullptr); + output_write_pending_event.store(nullptr); + } +} +#else +namespace +{ + void inject_thread_start_failure() + { + } + + void inject_handler_thread_start_failure() + { + } +} +#endif TwoWayPipeMessageIPC::TwoWayPipeMessageIPC( std::wstring _input_pipe_name, @@ -18,6 +180,7 @@ TwoWayPipeMessageIPC::TwoWayPipeMessageIPC( TwoWayPipeMessageIPC::~TwoWayPipeMessageIPC() { + impl->end(); delete impl; } @@ -58,66 +221,166 @@ void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::send(std::wstring msg) void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::start(HANDLE _restricted_pipe_token) { + std::scoped_lock lock(lifecycle_mutex); + if (lifecycle_state != LifecycleState::NotStarted) + { + return; + } + // Legacy overload = no caller authentication: explicitly clear any previously-set policy so this // path can never inherit a policy from a prior parameterized start on the same instance. caller_policy = {}; - 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); + 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::start(HANDLE _restricted_pipe_token, const interop_auth::CallerPolicy& _caller_policy) { + std::scoped_lock lock(lifecycle_mutex); + if (lifecycle_state != LifecycleState::NotStarted) + { + return; + } + // Start threads inline (do not chain into the legacy overload, which would clear the policy). caller_policy = _caller_policy; - 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); + 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. @@ -127,45 +390,96 @@ void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::send_pipe_message(std::wstr return; } - // All pipe instances are busy, so wait for 20 seconds. - - if (!WaitNamedPipe(lpszPipename, 20000)) + // Use short waits so end() can promptly join the output thread instead of waiting for a + // long unavailable-pipe timeout. +#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; + // Begin the overlapped write while holding the same mutex end() uses for + // cancellation. This closes the check-to-write race where shutdown could + // otherwise cancel before the write was issued. + 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; + } +#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS + if (const HANDLE pending_event = output_write_pending_event.load()) + { + SetEvent(pending_event); + } +#endif + 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) @@ -187,6 +501,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. @@ -238,13 +553,13 @@ BOOL TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::GetLogonSID(HANDLE hToken, if (!CopySid(dwLength, *ppsid, ptg->Groups[dwIndex].Sid)) { HeapFree(GetProcessHeap(), 0, static_cast(*ppsid)); + *ppsid = nullptr; goto Cleanup; } + bSuccess = TRUE; break; } - bSuccess = TRUE; - Cleanup: // Free the buffer for the token groups. @@ -261,71 +576,138 @@ VOID TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::FreeLogonSID(PSID* ppsid) HeapFree(GetProcessHeap(), 0, static_cast(*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(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(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) + { + // A non-elevated server has no identity distinct from its same-user clients. Retain its + // existing multi-instance behavior with an explicit DACL; elevated Runner servers use the + // Administrators-owned path below, which is the security boundary this transport needs. + DWORD token_user_size = 0; + GetTokenInformation(token, TokenUser, nullptr, 0, &token_user_size); + auto* token_user = static_cast(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(new_dacl)); -Lclean_sd: - LocalFree(static_cast(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; + } + + // The elevated server identity can change the DACL or create later instances. The + // medium-integrity client receives the exact data/attribute rights it needs, never default or + // creator-owner rights. + 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() @@ -364,124 +746,305 @@ 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& 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; } + bool accepted = !closed.load(); + // Authenticate the connecting client before reading/queuing anything. Fail-closed: an unauthenticated // caller gets no dispatch. When the policy is disabled (managed server / tests) this is a no-op. - if (caller_policy.enabled) + if (accepted && caller_policy.enabled) { const interop_auth::AuthResult auth = interop_auth::AuthenticateClient(input_pipe_handle, caller_policy, caller_cache); if (!auth.accepted) { - FlushFileBuffers(input_pipe_handle); - DisconnectNamedPipe(input_pipe_handle); - CloseHandle(input_pipe_handle); - return; + accepted = false; } } - constexpr DWORD readBlockBytes = BUFSIZE; - std::wstring message; - size_t iBlock = 0; - message.reserve(BUFSIZE); - bool ok; - do + if (accepted && !closed.load()) { - 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) + constexpr DWORD readBlockBytes = BUFSIZE; + std::wstring message; + size_t iBlock = 0; + message.reserve(BUFSIZE); + bool message_read = false; + do { - break; + constexpr size_t charsPerBlock = readBlockBytes / sizeof(message[0]); + message.resize(message.size() + charsPerBlock); + DWORD bytesRead = 0; + message_read = ReadFile( + input_pipe_handle, + // Read the message directly into the string block by block while resizing it. + message.data() + iBlock * charsPerBlock, + readBlockBytes, + &bytesRead, + nullptr); + + if (!message_read && GetLastError() != ERROR_MORE_DATA) + { + break; + } + iBlock++; + } while (!message_read); + + if (message_read && !closed.load()) + { + // Trim the message's buffer. + 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)); + + // Flush the pipe to allow the client to read the pipe's contents before disconnecting. + FlushFileBuffers(input_pipe_handle); } - iBlock++; - } while (!ok); - // trim the message's buffer - const auto nullCharPos = message.find_last_not_of(L'\0'); - if (nullCharPos != std::wstring::npos) + } + finish_connection_handler(handler); +#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS + if (const HANDLE completed_event = handler_completed_event.load()) { - message.resize(nullCharPos + 1); + SetEvent(completed_event); + if (const HANDLE allow_return_event = handler_allow_return_event.load()) + { + WaitForSingleObject(allow_return_event, 10'000); + } + } +#endif +} + +void TwoWayPipeMessageIPC::TwoWayPipeMessageIPCImpl::finish_connection_handler(const std::shared_ptr& 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(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 + { +#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS + inject_handler_thread_start_failure(); +#endif + 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> 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> 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) + // Create the first instance with FILE_FLAG_FIRST_PIPE_INSTANCE so that CreateNamedPipe + // fails fast if a pipe with this name already exists (for example a leftover instance + // from a previous run or another process), making this server the sole owner of the + // pipe name instead of silently sharing it. The flag is only valid on the first + // instance; subsequent instances must omit it. + auto create_listener = [&](bool first_instance) { +#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS + if (!first_instance) + { + if (const HANDLE reached_event = before_replacement_listener_event.load()) + { + SetEvent(reached_event); + if (const HANDLE allow_creation_event = allow_replacement_listener_event.load()) + { + WaitForSingleObject(allow_creation_event, 10'000); + } + } + } +#endif + 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_REJECT_REMOTE_CLIENTS, - 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 + + // Claim the replacement listener before giving the accepted instance to its handler. This + // keeps at least one secured instance alive even if a rejected client closes immediately. + 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); + } } + +#ifdef TWO_WAY_PIPE_MESSAGE_IPC_TESTS + if (replacement.valid()) + { + if (const HANDLE reached_event = after_replacement_listener_event.load()) + { + SetEvent(reached_event); + if (const HANDLE allow_continue_event = allow_after_replacement_listener_event.load()) + { + WaitForSingleObject(allow_continue_event, 10'000); + } + } + } +#endif + 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(); diff --git a/src/common/interop/two_way_pipe_message_ipc.h b/src/common/interop/two_way_pipe_message_ipc.h index 96e467ff7f..48778bcd05 100644 --- a/src/common/interop/two_way_pipe_message_ipc.h +++ b/src/common/interop/two_way_pipe_message_ipc.h @@ -1,6 +1,28 @@ #pragma once #include #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: diff --git a/src/common/interop/two_way_pipe_message_ipc_impl.h b/src/common/interop/two_way_pipe_message_ipc_impl.h index 495bb95758..766902d55c 100644 --- a/src/common/interop/two_way_pipe_message_ipc_impl.h +++ b/src/common/interop/two_way_pipe_message_ipc_impl.h @@ -1,10 +1,14 @@ #pragma once +#include +#include +#include +#include #include #include "async_message_queue.h" #include #include #include -#include +#include #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> 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& handler); + void finish_connection_handler(const std::shared_ptr& 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(); }; diff --git a/src/modules/Workspaces/WorkspacesLib/two_way_pipe_message_ipc.cpp b/src/modules/Workspaces/WorkspacesLib/two_way_pipe_message_ipc.cpp index 40b2f1dbe5..1a760b1a31 100644 --- a/src/modules/Workspaces/WorkspacesLib/two_way_pipe_message_ipc.cpp +++ b/src/modules/Workspaces/WorkspacesLib/two_way_pipe_message_ipc.cpp @@ -2,9 +2,92 @@ #include +#include #include +#include 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 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(*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(*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(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(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(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(new_dacl)); -Lclean_sd: - LocalFree(static_cast(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& 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& 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(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> 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> 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();