From 55b520235fd06e6d3d783efabc8b6679066a1107 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Fri, 15 May 2026 18:58:49 +0800 Subject: [PATCH] usbip: harden darwin response and stale-state handling --- service/usbip/client_darwin.go | 48 ++++++++++---- service/usbip/darwin_integration_test.go | 76 ++++++++++++++++++++++ service/usbip/data_protocol.go | 7 +- service/usbip/export_ledger.go | 2 +- service/usbip/export_ledger_darwin_test.go | 56 ++++++++++++++++ service/usbip/host.go | 6 +- service/usbip/host_darwin.go | 17 +++-- service/usbip/usbhost_darwin.go | 9 ++- 8 files changed, 197 insertions(+), 24 deletions(-) create mode 100644 service/usbip/export_ledger_darwin_test.go diff --git a/service/usbip/client_darwin.go b/service/usbip/client_darwin.go index 749d98342..fdaa2c4af 100644 --- a/service/usbip/client_darwin.go +++ b/service/usbip/client_darwin.go @@ -155,16 +155,22 @@ func (c *darwinVirtualController) readLoop() { } switch header.Command { case RetSubmit: - c.pendingAccess.Lock() - pending, ok := c.pending[header.SeqNum] - c.pendingAccess.Unlock() - payloadDirection := header.Direction - if ok { - payloadDirection = pending.direction + pending, ok := c.pendingSubmit(header.SeqNum) + if !ok { + err = E.New("unexpected RET_SUBMIT seq ", header.SeqNum) + c.logger.Debug(err) + c.runErr = err + c.requestClose() + c.failPending() + return } - response, err := ReadSubmitResponseBody(c.conn, header, payloadDirection) + response, err := ReadSubmitResponseBody(c.conn, header, pending.direction) if err != nil { c.logger.Debug("read RET_SUBMIT: ", err) + if !E.IsClosedOrCanceled(err) { + c.runErr = err + } + c.requestClose() c.failPending() return } @@ -173,11 +179,18 @@ func (c *darwinVirtualController) readLoop() { _, err := ReadUnlinkResponseBody(c.conn, header) if err != nil { c.logger.Debug("read RET_UNLINK: ", err) + if !E.IsClosedOrCanceled(err) { + c.runErr = err + } + c.requestClose() c.failPending() return } default: - c.logger.Debug(fmt.Sprintf("unexpected USB/IP response 0x%08x", header.Command)) + err = E.New(fmt.Sprintf("unexpected USB/IP response 0x%08x", header.Command)) + c.logger.Debug(err) + c.runErr = err + c.requestClose() c.failPending() return } @@ -588,15 +601,11 @@ func (c *darwinVirtualController) sendSubmit(command SubmitCommand) (SubmitRespo c.pendingAccess.Lock() c.pending[seq] = darwinPendingSubmit{direction: command.Header.Direction, reply: reply} c.pendingAccess.Unlock() - defer func() { - c.pendingAccess.Lock() - delete(c.pending, seq) - c.pendingAccess.Unlock() - }() c.writeAccess.Lock() err := WriteSubmitCommand(c.conn, command) c.writeAccess.Unlock() if err != nil { + c.removePendingSubmit(seq) return SubmitResponse{}, err } select { @@ -610,6 +619,19 @@ func (c *darwinVirtualController) sendSubmit(command SubmitCommand) (SubmitRespo } } +func (c *darwinVirtualController) pendingSubmit(seq uint32) (darwinPendingSubmit, bool) { + c.pendingAccess.Lock() + defer c.pendingAccess.Unlock() + pending, ok := c.pending[seq] + return pending, ok +} + +func (c *darwinVirtualController) removePendingSubmit(seq uint32) { + c.pendingAccess.Lock() + delete(c.pending, seq) + c.pendingAccess.Unlock() +} + func (c *darwinVirtualController) deliverSubmit(response SubmitResponse) { c.pendingAccess.Lock() pending, ok := c.pending[response.Header.SeqNum] diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index b829df41a..33174c9a4 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -196,6 +196,82 @@ func TestDarwinUSBHostDeviceWatcherSmoke(t *testing.T) { watcher.Close() } +func TestDarwinControllerReadsINRetSubmitUsingPendingDirection(t *testing.T) { + t.Parallel() + + clientConn, serverConn := net.Pipe() + defer serverConn.Close() + + controller := newDarwinVirtualController(context.Background(), newTestLogger(t), clientConn, darwinFakeDeviceEntry().Info) + go controller.readLoop() + defer func() { + _ = controller.Close() + <-controller.Done() + }() + + reply := make(chan SubmitResponse, 1) + controller.pendingAccess.Lock() + controller.pending[1] = darwinPendingSubmit{ + direction: USBIPDirIn, + reply: reply, + } + controller.pendingAccess.Unlock() + + entry := darwinFakeDeviceEntry() + expectedPayload := []byte{0xde, 0xad, 0xbe, 0xef} + err := WriteSubmitResponse(serverConn, SubmitResponse{ + Header: DataHeader{ + Command: RetSubmit, + SeqNum: 1, + DevID: entry.Info.DevID(), + Direction: USBIPDirIn, + Endpoint: 1, + }, + ActualLength: int32(len(expectedPayload)), + NumberOfPackets: nonIsoPacketCount, + Buffer: expectedPayload, + }) + require.NoError(t, err) + + select { + case response, ok := <-reply: + require.True(t, ok) + require.Equal(t, expectedPayload, response.Buffer) + require.Zero(t, response.Header.Direction) + case <-time.After(time.Second): + t.Fatal("timed out waiting for RET_SUBMIT") + } + + controller.pendingAccess.Lock() + _, found := controller.pending[1] + controller.pendingAccess.Unlock() + require.False(t, found) +} + +func TestDarwinControllerRejectsUnknownRetSubmitImmediately(t *testing.T) { + t.Parallel() + + clientConn, serverConn := net.Pipe() + defer serverConn.Close() + + controller := newDarwinVirtualController(context.Background(), newTestLogger(t), clientConn, darwinFakeDeviceEntry().Info) + go controller.readLoop() + + err := writeDataHeader(serverConn, DataHeader{ + Command: RetSubmit, + SeqNum: 99, + }) + require.NoError(t, err) + + select { + case <-controller.Done(): + case <-time.After(time.Second): + t.Fatal("timed out waiting for controller to reject unexpected RET_SUBMIT") + } + + require.ErrorContains(t, controller.Err(), "unexpected RET_SUBMIT seq 99") +} + func startDarwinFakeUSBIPServer(t *testing.T) *darwinFakeUSBIPServer { t.Helper() diff --git a/service/usbip/data_protocol.go b/service/usbip/data_protocol.go index 5a2cd1d5c..5efcf385c 100644 --- a/service/usbip/data_protocol.go +++ b/service/usbip/data_protocol.go @@ -112,7 +112,10 @@ func ReadSubmitCommandBody(r io.Reader, header DataHeader) (SubmitCommand, error return command, nil } -func ReadSubmitResponseBody(r io.Reader, header DataHeader, payloadDirection uint32) (SubmitResponse, error) { +// ReadSubmitResponseBody decodes a RET_SUBMIT body. requestDirection must be +// the original CMD_SUBMIT direction, because the USB/IP response header zeroes +// direction on the wire. +func ReadSubmitResponseBody(r io.Reader, header DataHeader, requestDirection uint32) (SubmitResponse, error) { var raw [28]byte if _, err := io.ReadFull(r, raw[:]); err != nil { return SubmitResponse{}, err @@ -127,7 +130,7 @@ func ReadSubmitResponseBody(r io.Reader, header DataHeader, payloadDirection uin } copy(response.Setup[:], raw[20:28]) bufferLength := max(response.ActualLength, 0) - buffer, isoPackets, err := readUSBIPPayload(r, payloadDirection, bufferLength, response.NumberOfPackets, false) + buffer, isoPackets, err := readUSBIPPayload(r, requestDirection, bufferLength, response.NumberOfPackets, false) if err != nil { return SubmitResponse{}, err } diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go index a451097fc..cc69d6149 100644 --- a/service/usbip/export_ledger.go +++ b/service/usbip/export_ledger.go @@ -516,7 +516,7 @@ func (l *exportLedger) snapshotDeviceState(ctx context.Context) []DeviceInfoV2 { out := make([]DeviceInfoV2, 0, len(entries)) for _, e := range entries { snapshot := e.export.Snapshot(ctx, e.busy) - if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.IDVendor == 0 { + if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.BusIDString() == "" { continue } out = append(out, deviceInfoV2FromEntry(snapshot.Entry, snapshot.Backend, snapshot.StableID, snapshot.State, snapshot.RawStatus, snapshot.StatusReason)) diff --git a/service/usbip/export_ledger_darwin_test.go b/service/usbip/export_ledger_darwin_test.go new file mode 100644 index 000000000..ace9e5fee --- /dev/null +++ b/service/usbip/export_ledger_darwin_test.go @@ -0,0 +1,56 @@ +//go:build darwin && cgo + +package usbip + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) { + t.Parallel() + + ctx := context.Background() + ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) + entry := darwinFakeDeviceEntry() + export := &darwinExport{ + busid: entry.Info.BusIDString(), + registryID: 0x1234, + entry: entry, + } + + ledger.ApplyHostSnapshot(map[string]Export{export.busid: export}, nil) + ledger.SeedBroadcastState(ctx) + + sub, _ := ledger.Subscribe(ctx, nil, controlCapabilities) + select { + case <-sub.send: + case <-time.After(time.Second): + t.Fatal("timed out waiting for initial snapshot") + } + + export.stale = true + export.pendingRegistryID = 0x5678 + + if !ledger.BroadcastIfChanged(ctx) { + t.Fatal("expected stale darwin export to broadcast an update") + } + + select { + case message := <-sub.send: + require.Equal(t, controlFrameDeviceDelta, message.Frame.Type) + + var delta controlDeviceDelta + require.NoError(t, unmarshalControlPayload(message.Payload, &delta)) + require.Empty(t, delta.Removed) + require.Len(t, delta.Updated, 1) + require.Equal(t, export.busid, delta.Updated[0].BusID) + require.Equal(t, deviceStateUnavailable, delta.Updated[0].State) + require.Equal(t, "device replaced", delta.Updated[0].StatusReason) + case <-time.After(time.Second): + t.Fatal("timed out waiting for unavailable update") + } +} diff --git a/service/usbip/host.go b/service/usbip/host.go index 786b6f44a..42601c238 100644 --- a/service/usbip/host.go +++ b/service/usbip/host.go @@ -42,8 +42,10 @@ type Export interface { } // ExportSnapshot Backend and StableID fields are populated -// unconditionally so the caller can build an unavailable record even -// when Entry could not be read. +// unconditionally. Unavailable snapshots should keep a cached Entry, +// including BusID, so the caller can broadcast a state transition +// instead of removing the device outright; snapshots without a BusID +// are treated as non-broadcastable. type ExportSnapshot struct { Entry DeviceEntry Backend string diff --git a/service/usbip/host_darwin.go b/service/usbip/host_darwin.go index 5f83628d9..9d377cfb6 100644 --- a/service/usbip/host_darwin.go +++ b/service/usbip/host_darwin.go @@ -286,13 +286,22 @@ func (e *darwinExport) LeaseIdentity() ExportLeaseIdentity { return ExportLeaseIdentity(fmt.Sprintf("darwin:%016x", e.registryID)) } +func (e *darwinExport) staleReason() string { + if e.pendingRegistryID != 0 { + return "device replaced" + } + return "capture released" +} + func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot { stableID := fmt.Sprintf("darwin-registry:%016x", e.registryID) if e.stale { return ExportSnapshot{ - Backend: backendIDDarwinIOKit, - StableID: stableID, - State: deviceStateUnavailable, + Entry: e.entry, + Backend: backendIDDarwinIOKit, + StableID: stableID, + State: deviceStateUnavailable, + StatusReason: e.staleReason(), } } state := deviceStateAvailable @@ -309,7 +318,7 @@ func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot { func (e *darwinExport) LeaseCheck(ctx context.Context) (bool, string) { if e.stale { - return false, "capture released" + return false, e.staleReason() } return true, "" } diff --git a/service/usbip/usbhost_darwin.go b/service/usbip/usbhost_darwin.go index 22ec9b1ec..de28dbbe0 100644 --- a/service/usbip/usbhost_darwin.go +++ b/service/usbip/usbhost_darwin.go @@ -548,6 +548,9 @@ func darwinIOReturnToUSBIPStatus(status int32) int32 { } } +// darwinUSBIPStatusToCIStatus maps USB/IP transfer completion status to the +// corresponding IOUSBHostCI completion status. This is only used for real +// transfer completion, not for EndpointPause-driven state machine events. func darwinUSBIPStatusToCIStatus(status int32) int { if status == 0 { return int(C.IOUSBHostCIMessageStatusSuccess) @@ -555,6 +558,8 @@ func darwinUSBIPStatusToCIStatus(status int32) int { switch -status { case int32(unix.EPIPE): return int(C.IOUSBHostCIMessageStatusStallError) + case int32(unix.ENODEV), int32(unix.ECONNRESET): + return int(C.IOUSBHostCIMessageStatusOffline) case int32(unix.ETIMEDOUT): return int(C.IOUSBHostCIMessageStatusTimeout) case int32(unix.ENOMEM): @@ -563,8 +568,8 @@ func darwinUSBIPStatusToCIStatus(status int32) int { return int(C.IOUSBHostCIMessageStatusBadArgument) case int32(unix.EPERM): return int(C.IOUSBHostCIMessageStatusNotPermitted) - case int32(unix.ECONNRESET): - return int(C.IOUSBHostCIMessageStatusEndpointStopped) + case int32(unix.EOVERFLOW): + return int(C.IOUSBHostCIMessageStatusOverrunError) default: return int(C.IOUSBHostCIMessageStatusError) }