fix(ds5): 加固 sidecar owner 鉴权与管道传输健壮性

docs/windows_dualsense_component_lifecycle.md

@@ -358,6 +358,8 @@ shutdown(owner_token)
 - `attach`、`update_input`、`subscribe_output` 和 `get_status` 只接受当前连接 owner；Core 持有 owner token，GUI 测试使用独立、低权限 test token。
 - Sidecar 拒绝非所有者 detach 和 shutdown；连接断开会清理该 owner 创建的全部设备。
 - 输出报告和音频数据使用有界队列；控制消息不得被高频数据饿死。
+- 已实现的 owner 校验（v1）：管道 ACL 限定当前用户 + Sidecar 在连接建立时校验客户端进程的提权状态，非提权客户端拒绝并断开、继续等待真正的 owner（不因被抢连而退出，避免单次抢连导致该会话分配失败）；同用户非提权进程即使抢到单实例管道也无法驱动 elevated Sidecar。GUI 低权限 test token 仍属后续工作。
+- 已实现的停滞保护（v1）：Core 对数据面写操作设置 5 秒停滞上限；写停滞会取消 reader 的挂起读取并进入既有的单次恢复路径，sidecar 读循环阻塞不再冻结 Sunshine 输入线程。
 
 高频四声道音频数据不应经 Tauri 或 JSON 传输。后续实现使用共享内存环形缓冲区或专用本地数据通道；Named Pipe 只负责控制和状态。
 
@@ -492,7 +494,7 @@ Core 维护引用计数，按 session ID 管理多客户端。第一阶段可以
 3. 安装器只操作固定组件根目录，删除前解析并验证绝对路径仍位于该根目录。
 4. Sidecar 启动路径必须来自已验证 active manifest。
 5. 不把管理员权限传给常驻 Sidecar；提权 helper 只执行单个、结构化的驱动操作。
-6. Named Pipe ACL 仅允许当前用户、Sunshine 服务身份和管理员访问。
+6. Named Pipe ACL 仅允许当前用户、Sunshine 服务身份和管理员访问；Sidecar 在连接建立时校验客户端进程已提权，拒绝非提权连接。
 7. 不按可执行文件名全局终止进程。
 8. 每个异步操作只有一个 writer，并有 operation ID、取消状态和可恢复 snapshot。
 9. GUI 收到的错误文本视为不可信数据，显示时转义；用户文案由稳定错误码映射。

src/platform/windows/ds5/ds5_sidecar_client.cpp

@@ -40,6 +40,8 @@ namespace platf::ds5 {
     constexpr std::uint16_t VERSION = 1;
     constexpr std::size_t HEADER_SIZE = 16;
     constexpr std::uint32_t MAX_PAYLOAD = 1024 * 1024;
+    // The sidecar protocol identifies devices with a single byte.
+    static_assert(platf::MAX_GAMEPADS <= 256, "DS5 device ids must fit the wire format");
 
     enum class message_e: std::uint16_t {
       hello = 1,
@@ -93,7 +95,13 @@ namespace platf::ds5 {
       p[3] = static_cast<std::uint8_t>(value >> 24);
     }
 
-    bool transfer_exact(HANDLE pipe, HANDLE stop_event, void *buffer, std::size_t size, bool write) {
+    // A data-plane write that cannot drain within this bound means the sidecar
+    // stopped reading (for example a blocked HIDMaestro call on its read loop);
+    // failing the write lets the caller bail out instead of freezing.
+    constexpr DWORD write_stall_timeout_ms = 5000;
+
+    bool transfer_exact(HANDLE pipe, HANDLE stop_event, void *buffer, std::size_t size, bool write,
+                        DWORD wait_timeout = INFINITE) {
       std::size_t offset = 0;
       while (offset < size) {
         if (WaitForSingleObject(stop_event, 0) == WAIT_OBJECT_0) {
@@ -131,7 +139,7 @@ namespace platf::ds5 {
 #endif
           const std::array waits { overlapped.hEvent, stop_event };
           const auto wait_result = WaitForMultipleObjects(
-            static_cast<DWORD>(waits.size()), waits.data(), FALSE, INFINITE);
+            static_cast<DWORD>(waits.size()), waits.data(), FALSE, wait_timeout);
           if (wait_result == WAIT_OBJECT_0) {
             completed = GetOverlappedResult(pipe, &overlapped, &count, FALSE) != FALSE;
           }
@@ -159,7 +167,8 @@ namespace platf::ds5 {
 
     bool write_exact(HANDLE pipe, HANDLE stop_event, std::span<const std::uint8_t> source) {
       return transfer_exact(
-        pipe, stop_event, const_cast<std::uint8_t *>(source.data()), source.size(), true);
+        pipe, stop_event, const_cast<std::uint8_t *>(source.data()), source.size(), true,
+        write_stall_timeout_ms);
     }
 
 #ifndef SUNSHINE_DS5_SIDECAR_TEST_HOOK
@@ -287,7 +296,17 @@ namespace platf::ds5 {
       write_u32(frame.data() + 12, request_id);
       std::copy(payload.begin(), payload.end(), frame.begin() + HEADER_SIZE);
       std::lock_guard lock(write_mutex);
-      return !stopping && pipe != INVALID_HANDLE_VALUE && write_exact(pipe, stop_event, frame);
+      if (stopping || pipe == INVALID_HANDLE_VALUE) {
+        return false;
+      }
+      if (write_exact(pipe, stop_event, frame)) {
+        return true;
+      }
+      // The write stalled or the pipe broke: cancel the reader's pending read
+      // so read_loop's recovery path tears the transport down, instead of
+      // letting later senders block on a pipe that will never drain.
+      CancelIoEx(pipe, nullptr);
+      return false;
     }
 
     bool receive(message_t &message) {
@@ -306,22 +325,81 @@ namespace platf::ds5 {
       return read_exact(pipe, stop_event, message.payload);
     }
 
+    void dispatch(message_t &message) {
+      const auto &p = message.payload;
+      const auto owned_index = global_index.load();
+      switch (message.type) {
+        case message_e::rumble:
+          if (p.size() == 6 && p[0] == owned_index) {
+            feedback_queue->raise(gamepad_feedback_msg_t::make_rumble(
+              p[1], read_u16(p.data() + 2), read_u16(p.data() + 4)));
+          }
+          break;
+        case message_e::adaptive_triggers:
+          if (p.size() == 26 && p[0] == owned_index) {
+            std::array<std::uint8_t, 10> left, right;
+            std::copy_n(p.data() + 6, 10, left.begin());
+            std::copy_n(p.data() + 16, 10, right.begin());
+            feedback_queue->raise(gamepad_feedback_msg_t::make_adaptive_triggers(
+              p[1], p[2], p[3], p[4], left, right));
+          }
+          break;
+        case message_e::led:
+          if (p.size() == 5 && p[0] == owned_index) {
+            feedback_queue->raise(gamepad_feedback_msg_t::make_rgb_led(p[1], p[2], p[3], p[4]));
+          }
+          break;
+        case message_e::haptics_pcm:
+          if (p.size() >= 24 && p[0] == owned_index) {
+            const auto frames = read_u16(p.data() + 4);
+            const auto pcm_size = static_cast<std::size_t>(frames) * 4;
+            if (p[3] == 2 && p[6] == 16 && read_u32(p.data() + 20) == 48000 &&
+                frames <= 240 && p.size() == 24 + pcm_size) {
+              feedback_queue->raise(gamepad_feedback_msg_t::make_ds5_haptics_pcm(
+                p[1], p[2], frames, read_u32(p.data() + 8), read_u64(p.data() + 12),
+                p.data() + 24, pcm_size));
+            }
+          }
+          break;
+        case message_e::error:
+          BOOST_LOG(warning) << "DualSense sidecar reported an asynchronous error"sv;
+          break;
+        default:
+          break;
+      }
+    }
+
     bool transact(message_e request_type, std::span<const std::uint8_t> payload,
                   message_e reply_type, message_t &reply) {
       const auto request_id = next_request_id++;
-      if (!send(request_type, request_id, payload) || !receive(reply)) {
+      if (!send(request_type, request_id, payload)) {
         return false;
       }
-      if (reply.type == message_e::error) {
-        std::string reason;
-        if (reply.payload.size() >= 8) {
-          const auto length = std::min<std::size_t>(read_u32(reply.payload.data() + 4), reply.payload.size() - 8);
-          reason.assign(reinterpret_cast<const char *>(reply.payload.data() + 8), length);
+      while (!stopping) {
+        // Rumble, LEDs and adaptive triggers share the control channel and are
+        // emitted from a different sidecar thread, so they can legitimately
+        // arrive ahead of the reply. Dispatch them instead of misreading the
+        // first message as the reply.
+        message_t message;
+        if (!receive(message)) {
+          return false;
         }
-        BOOST_LOG(error) << "DualSense sidecar rejected request: "sv << reason;
-        return false;
+        if (message.request_id == request_id && message.type == reply_type) {
+          reply = std::move(message);
+          return true;
+        }
+        if (message.type == message_e::error && message.request_id == request_id) {
+          std::string reason;
+          if (message.payload.size() >= 8) {
+            const auto length = std::min<std::size_t>(read_u32(message.payload.data() + 4), message.payload.size() - 8);
+            reason.assign(reinterpret_cast<const char *>(message.payload.data() + 8), length);
+          }
+          BOOST_LOG(error) << "DualSense sidecar rejected request: "sv << reason;
+          return false;
+        }
+        dispatch(message);
       }
-      return reply.type == reply_type && reply.request_id == request_id;
+      return false;
     }
 
     bool launch_and_connect() {
@@ -417,6 +495,9 @@ namespace platf::ds5 {
         static_cast<std::uint8_t>(audio_haptics ? 1 : 0),
         0,
       };
+      // Claim ownership before the transaction so feedback interleaved ahead
+      // of the attach reply is routed to the queue instead of dropped.
+      global_index = id.globalIndex;
       if (!transact(message_e::attach, attach_payload, message_e::attach_reply, reply) ||
           reply.payload.size() != 8 || reply.payload[0] != attach_payload[0]) {
         return false;
@@ -426,7 +507,6 @@ namespace platf::ds5 {
         return false;
       }
 
-      global_index = id.globalIndex;
       client_index = id.clientRelativeIndex;
       audio_haptics_requested = audio_haptics;
       online = true;
@@ -477,47 +557,7 @@ namespace platf::ds5 {
       while (!stopping) {
         message_t message;
         while (!stopping && receive(message)) {
-          const auto &p = message.payload;
-          const auto owned_index = global_index.load();
-          switch (message.type) {
-            case message_e::rumble:
-              if (p.size() == 6 && p[0] == owned_index) {
-                feedback_queue->raise(gamepad_feedback_msg_t::make_rumble(
-                  p[1], read_u16(p.data() + 2), read_u16(p.data() + 4)));
-              }
-              break;
-            case message_e::adaptive_triggers:
-              if (p.size() == 26 && p[0] == owned_index) {
-                std::array<std::uint8_t, 10> left, right;
-                std::copy_n(p.data() + 6, 10, left.begin());
-                std::copy_n(p.data() + 16, 10, right.begin());
-                feedback_queue->raise(gamepad_feedback_msg_t::make_adaptive_triggers(
-                  p[1], p[2], p[3], p[4], left, right));
-              }
-              break;
-            case message_e::led:
-              if (p.size() == 5 && p[0] == owned_index) {
-                feedback_queue->raise(gamepad_feedback_msg_t::make_rgb_led(p[1], p[2], p[3], p[4]));
-              }
-              break;
-            case message_e::haptics_pcm:
-              if (p.size() >= 24 && p[0] == owned_index) {
-                const auto frames = read_u16(p.data() + 4);
-                const auto pcm_size = static_cast<std::size_t>(frames) * 4;
-                if (p[3] == 2 && p[6] == 16 && read_u32(p.data() + 20) == 48000 &&
-                    frames <= 240 && p.size() == 24 + pcm_size) {
-                  feedback_queue->raise(gamepad_feedback_msg_t::make_ds5_haptics_pcm(
-                    p[1], p[2], frames, read_u32(p.data() + 8), read_u64(p.data() + 12),
-                    p.data() + 24, pcm_size));
-                }
-              }
-              break;
-            case message_e::error:
-              BOOST_LOG(warning) << "DualSense sidecar reported an asynchronous error"sv;
-              break;
-            default:
-              break;
-          }
+          dispatch(message);
         }
         online = false;
         if (stopping) {

tests/tools/ds5_fake_sidecar.cpp

@@ -77,6 +77,9 @@ int main(int argc, char **argv) {
   const auto continue_name = L"Local\\sunshine-ds5-test-continue-" + event_suffix;
   const auto continue_event = OpenEventW(SYNCHRONIZE, FALSE, continue_name.c_str());
   if (!continue_event) return 2;
+  // The test process opts this peer into emitting async feedback ahead of the
+  // attach reply, exercising the Core client's transaction multiplexing.
+  const auto interleave = GetEnvironmentVariableW(L"SUNSHINE_DS5_TEST_INTERLEAVE", nullptr, 0) > 0;
   const auto crash_once_name = L"Local\\sunshine-ds5-test-crash-once-" + event_suffix;
   const auto crash_once_event = OpenEventW(SYNCHRONIZE | EVENT_MODIFY_STATE, FALSE,
                                            crash_once_name.c_str());
@@ -108,6 +111,11 @@ int main(int argc, char **argv) {
     if (type == 1) {
       if (!reply(pipe, 2, request_id, std::vector<std::uint8_t>(4))) break;
     } else if (type == 3 && payload.size() == 4) {
+      if (interleave) {
+        std::vector<std::uint8_t> early(6);
+        early[0] = payload[0];
+        if (!reply(pipe, 101, 0, early)) break;
+      }
       std::vector<std::uint8_t> response(8);
       response[0] = payload[0];
       if (!reply(pipe, 4, request_id, response)) break;

tests/unit/platform/windows/test_ds5_sidecar_client.cpp

@@ -89,6 +89,41 @@ TEST(Ds5SidecarClientTests, AllocThenFreeCancelsBlockedReader) {
   EXPECT_LT(elapsed, std::chrono::seconds(2));
 }
 
+TEST(Ds5SidecarClientTests, AttachSurvivesInterleavedAsyncFeedback) {
+  config_scope_t restore_config;
+  config::input.ds5_enabled = true;
+  config::input.ds5_sidecar_path = SUNSHINE_DS5_FAKE_SIDECAR_PATH;
+
+  event_namespace_scope_t events(L"interleaved-feedback");
+  const auto continue_name = L"Local\\sunshine-ds5-test-continue-" + events.suffix;
+  const auto marker_name = L"Local\\sunshine-ds5-test-marker-" + events.suffix;
+  handle_scope_t continue_event(CreateEventW(nullptr, FALSE, FALSE, continue_name.c_str()));
+  handle_scope_t marker_event(CreateEventW(nullptr, FALSE, FALSE, marker_name.c_str()));
+  ASSERT_NE(continue_event.handle, nullptr);
+  ASSERT_NE(marker_event.handle, nullptr);
+  ASSERT_NE(SetEnvironmentVariableW(L"SUNSHINE_DS5_TEST_INTERLEAVE", L"1"), 0);
+
+  auto mail = std::make_shared<safe::mail_raw_t>();
+  auto feedback = mail->queue<platf::gamepad_feedback_msg_t>("ds5-interleave-test");
+  auto feedback_for_test = feedback;
+  platf::ds5::sidecar_client_t client;
+  // The fake peer emits an async rumble ahead of the attach reply. The
+  // transaction must dispatch it and still match the reply; before the
+  // multiplexing fix the rumble was misread as the reply and alloc failed.
+  EXPECT_EQ(client.alloc({ 0, 0 }, std::move(feedback), false), 0);
+  SetEnvironmentVariableW(L"SUNSHINE_DS5_TEST_INTERLEAVE", nullptr);
+  const auto early = feedback_for_test->pop(std::chrono::seconds(2));
+  ASSERT_TRUE(early);
+  EXPECT_EQ(early->type, platf::gamepad_feedback_e::rumble);
+
+  ASSERT_TRUE(SetEvent(continue_event.handle));
+  ASSERT_EQ(WaitForSingleObject(marker_event.handle, 2000), WAIT_OBJECT_0);
+  const auto marker = feedback_for_test->pop(std::chrono::seconds(2));
+  ASSERT_TRUE(marker);
+  EXPECT_EQ(marker->type, platf::gamepad_feedback_e::rumble);
+  client.free(0);
+}
+
 TEST(Ds5SidecarClientTests, RejectsCompositeAttachWithoutAudioEndpoint) {
   config_scope_t restore_config;
   config::input.ds5_enabled = true;

tools/sunshine-ds5-sidecar/ControllerSession.cs

@@ -42,6 +42,7 @@ internal sealed class ControllerSession : IDisposable
     private int _hapticsStreaming;
     private int _hapticsNeedsStart;
     private int _disposed;
+    private byte[] _audioResidual = Array.Empty<byte>();
 
     internal ControllerSession(byte deviceId,
                                byte clientControllerNumber,
@@ -224,17 +225,44 @@ private void OnOutputDecoded(object? sender, HMOutputDecodedEventArgs output)
 
     private void OnAudioStreamingChanged(object? sender, bool streaming)
     {
-        Interlocked.Exchange(ref _hapticsStreaming, streaming ? 1 : 0);
         if (streaming)
+        {
+            // Arm the start marker before publishing the streaming flag, or a
+            // frame racing this callback could be emitted mid-stream without
+            // the StreamStart marker the client resets on.
             Interlocked.Exchange(ref _hapticsNeedsStart, 1);
+            Interlocked.Exchange(ref _hapticsStreaming, 1);
+        }
         else
+        {
+            Interlocked.Exchange(ref _hapticsStreaming, 0);
+            // A stale sub-frame tail from the old stream must not splice into
+            // the first frame of the next stream.
+            _audioResidual = Array.Empty<byte>();
             EmitHaptics(ReadOnlySpan<byte>.Empty, 0, Protocol.HapticsFlags.StreamEnd);
+        }
     }
 
     private void OnAudioFrames(object? sender, ReadOnlyMemory<byte> pcm)
     {
-        var source = pcm.Span;
+        // Only the USB audio output thread raises this callback, so the
+        // residual carry is not guarded by a lock.
+        byte[] combined;
+        if (_audioResidual.Length == 0)
+        {
+            combined = pcm.ToArray();
+        }
+        else
+        {
+            combined = new byte[_audioResidual.Length + pcm.Length];
+            _audioResidual.AsSpan().CopyTo(combined);
+            pcm.Span.CopyTo(combined.AsSpan(_audioResidual.Length));
+        }
         var sourceFrameBytes = AudioInputChannels * BytesPerSample;
+        var usableBytes = combined.Length - combined.Length % sourceFrameBytes;
+        _audioResidual = usableBytes == combined.Length ? Array.Empty<byte>() : combined[usableBytes..];
+
+        var source = combined.AsSpan(0, usableBytes);
         var frameCount = source.Length / sourceFrameBytes;
         var offsetFrames = 0;
         while (offsetFrames < frameCount)