From 8c5996c6bc7d756c7c72c944cde8ea29ebfbd0a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Tue, 19 May 2026 15:48:50 +0800 Subject: [PATCH] usbip: drop import-lease subsystem The plain OP_REQ_IMPORT + TryReserveForImport path already delivers race-free exclusive access for every observable feature; the lease two-phase commit (OP_REQ_IMPORT_EXT, control lease frames, serverImportLease) only added identity-replacement detection and pre-import busy advertisement, neither of which is asserted by any feature test. TryReserveForImport now runs as a single critical section under inventoryAccess, closing the TOCTOU window the LeaseIdentity recheck used to guard. Export interface loses LeaseIdentity/LeaseCheck. controlPingLoop writes directly via writeControlMessage, dropping the clientControlSession holder and the controlSession/controlAccess fields on ClientService. Control protocol drops the ImportLease capability bit; controlExtensionCapabilities is now PayloadFrames | DeviceStateV2. --- service/usbip/client.go | 26 +- service/usbip/client_control.go | 136 -------- service/usbip/client_shared.go | 36 +-- service/usbip/control_protocol.go | 24 +- service/usbip/darwin_integration_test.go | 38 +-- service/usbip/export_ledger.go | 354 ++------------------- service/usbip/export_ledger_darwin_test.go | 2 +- service/usbip/export_ledger_test.go | 161 +--------- service/usbip/host.go | 4 - service/usbip/host_darwin.go | 11 - service/usbip/host_linux.go | 41 --- service/usbip/host_windows.go | 8 - service/usbip/protocol.go | 49 +-- service/usbip/server.go | 33 +- service/usbip/server_shared.go | 6 - 15 files changed, 46 insertions(+), 883 deletions(-) diff --git a/service/usbip/client.go b/service/usbip/client.go index 1366f8574..f4f00262c 100644 --- a/service/usbip/client.go +++ b/service/usbip/client.go @@ -32,9 +32,6 @@ type ClientService struct { assignedWorkers []*clientAssignedWorker allWorkers map[string]context.CancelFunc - controlAccess sync.Mutex - controlSession *clientControlSession - remoteAccess sync.Mutex remoteDevicesV2 map[string]DeviceInfoV2 } @@ -155,26 +152,9 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (Attach stopCloseOnCancel := closeConnOnContextDone(ctx, conn) defer stopCloseOnCancel() - lease, err := c.requestImportLease(ctx, busid) + err = WriteOpReqImport(conn, busid) if err != nil { - return nil, err - } - expectedReply := OpRepImport - if lease.Valid { - expectedReply = OpRepImportExt - err = WriteOpReqImportExt(conn, ImportExtRequest{ - BusID: busid, - LeaseID: lease.ID, - ClientNonce: lease.ClientNonce, - }) - if err != nil { - return nil, E.Cause(err, "write OP_REQ_IMPORT_EXT") - } - } else { - err = WriteOpReqImport(conn, busid) - if err != nil { - return nil, E.Cause(err, "write OP_REQ_IMPORT") - } + return nil, E.Cause(err, "write OP_REQ_IMPORT") } header, err := ReadOpHeader(conn) if err != nil { @@ -183,7 +163,7 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (Attach if header.Version != ProtocolVersion { return nil, E.New("unexpected reply version ", fmt.Sprintf("0x%04x", header.Version)) } - if header.Code != expectedReply { + if header.Code != OpRepImport { return nil, E.New("unexpected reply code ", fmt.Sprintf("0x%04x", header.Code)) } if header.Status != OpStatusOK { diff --git a/service/usbip/client_control.go b/service/usbip/client_control.go index e7549cf60..508553487 100644 --- a/service/usbip/client_control.go +++ b/service/usbip/client_control.go @@ -4,144 +4,8 @@ package usbip import ( "context" - "net" - "sync" - "time" - - E "github.com/sagernet/sing/common/exceptions" ) -var errClientControlSessionClosed = E.New("usbip control session closed") - -type clientImportLease struct { - Valid bool - ID uint64 - ClientNonce uint64 -} - -type clientControlSession struct { - conn net.Conn - capabilities uint32 - writeAccess sync.Mutex - access sync.Mutex - nextNonce uint64 - pending map[uint64]chan clientLeaseResult - closed bool -} - -type clientLeaseResult struct { - response controlLeaseResponse - err error -} - -func newClientControlSession(conn net.Conn, capabilities uint32) *clientControlSession { - return &clientControlSession{ - conn: conn, - capabilities: capabilities, - pending: make(map[uint64]chan clientLeaseResult), - } -} - -func (s *clientControlSession) writeControl(frame controlFrame, payload any) error { - s.writeAccess.Lock() - defer s.writeAccess.Unlock() - _ = s.conn.SetWriteDeadline(time.Now().Add(controlWriteTimeout)) - err := writeControlMessage(s.conn, frame, payload) - _ = s.conn.SetWriteDeadline(time.Time{}) - return err -} - -func (s *clientControlSession) requestLease(ctx context.Context, busid string) (controlLeaseResponse, error) { - s.access.Lock() - if s.closed { - s.access.Unlock() - return controlLeaseResponse{}, errClientControlSessionClosed - } - s.nextNonce++ - nonce := s.nextNonce - waiter := make(chan clientLeaseResult, 1) - s.pending[nonce] = waiter - s.access.Unlock() - - request := controlLeaseRequest{ - BusID: busid, - ClientNonce: nonce, - } - err := s.writeControl(controlFrame{ - Type: controlFrameLeaseRequest, - Version: controlProtocolVersion, - }, request) - if err != nil { - s.access.Lock() - delete(s.pending, nonce) - s.access.Unlock() - return controlLeaseResponse{}, err - } - - select { - case result := <-waiter: - return result.response, result.err - case <-ctx.Done(): - s.access.Lock() - delete(s.pending, nonce) - s.access.Unlock() - return controlLeaseResponse{}, ctx.Err() - } -} - -func (s *clientControlSession) deliverLeaseResponse(response controlLeaseResponse) bool { - s.access.Lock() - waiter, ok := s.pending[response.ClientNonce] - if ok { - delete(s.pending, response.ClientNonce) - } - s.access.Unlock() - if !ok { - return false - } - waiter <- clientLeaseResult{response: response} - close(waiter) - return true -} - -func (s *clientControlSession) closeWithError(err error) { - s.access.Lock() - if s.closed { - s.access.Unlock() - return - } - s.closed = true - pending := s.pending - s.pending = make(map[uint64]chan clientLeaseResult) - s.access.Unlock() - - for _, waiter := range pending { - waiter <- clientLeaseResult{err: err} - close(waiter) - } -} - -func (c *ClientService) requestImportLease(ctx context.Context, busid string) (clientImportLease, error) { - c.controlAccess.Lock() - session := c.controlSession - c.controlAccess.Unlock() - if session == nil || !supportsControlExtensions(session.capabilities) { - return clientImportLease{}, nil - } - response, err := session.requestLease(ctx, busid) - if err != nil { - return clientImportLease{}, E.Cause(err, "request import lease") - } - if response.ErrorCode != "" { - return clientImportLease{}, E.New("remote rejected import lease (", response.ErrorCode, ": ", response.ErrorMessage, ")") - } - return clientImportLease{ - Valid: true, - ID: response.LeaseID, - ClientNonce: response.ClientNonce, - }, nil -} - func (c *ClientService) applyControlDelta(delta controlDeviceDelta) { c.remoteAccess.Lock() if c.remoteDevicesV2 == nil { diff --git a/service/usbip/client_shared.go b/service/usbip/client_shared.go index 9aba6bd61..b4dd074bc 100644 --- a/service/usbip/client_shared.go +++ b/service/usbip/client_shared.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "net" "slices" "time" @@ -179,21 +180,8 @@ func (c *ClientService) runControlSession() error { _ = conn.SetWriteDeadline(time.Time{}) _ = conn.SetReadDeadline(time.Time{}) - session := newClientControlSession(conn, ack.Capabilities) extended := supportsControlExtensions(ack.Capabilities) - if extended { - c.controlAccess.Lock() - c.controlSession = session - c.controlAccess.Unlock() - defer func() { - c.controlAccess.Lock() - if c.controlSession == session { - c.controlSession = nil - } - c.controlAccess.Unlock() - session.closeWithError(errClientControlSessionClosed) - }() - } else { + if !extended { err = c.syncRemoteStateContext(c.ctx) if err != nil { return E.Cause(err, "initial devlist sync") @@ -201,7 +189,7 @@ func (c *ClientService) runControlSession() error { } pingDone := make(chan struct{}) - go c.controlPingLoop(session, pingDone) + go c.controlPingLoop(conn, pingDone) defer close(pingDone) lastSeq := ack.Sequence @@ -266,16 +254,6 @@ func (c *ClientService) runControlSession() error { } lastSeq = frame.Sequence c.applyControlDelta(delta) - case controlFrameLeaseResponse: - if !extended { - return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type) - } - var response controlLeaseResponse - err = unmarshalControlPayload(message.Payload, &response) - if err != nil { - return E.Cause(errImmediateReconnect, "read lease response: ", err) - } - session.deliverLeaseResponse(response) case controlFramePong: default: return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type) @@ -283,7 +261,7 @@ func (c *ClientService) runControlSession() error { } } -func (c *ClientService) controlPingLoop(session *clientControlSession, done <-chan struct{}) { +func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) { ticker := time.NewTicker(controlPingInterval) defer ticker.Stop() for { @@ -293,12 +271,14 @@ func (c *ClientService) controlPingLoop(session *clientControlSession, done <-ch case <-done: return case <-ticker.C: - err := session.writeControl(controlFrame{ + _ = conn.SetWriteDeadline(time.Now().Add(controlWriteTimeout)) + err := writeControlMessage(conn, controlFrame{ Type: controlFramePing, Version: controlProtocolVersion, }, nil) + _ = conn.SetWriteDeadline(time.Time{}) if err != nil { - _ = session.conn.Close() + _ = conn.Close() return } } diff --git a/service/usbip/control_protocol.go b/service/usbip/control_protocol.go index e0da681bd..43dd19bcb 100644 --- a/service/usbip/control_protocol.go +++ b/service/usbip/control_protocol.go @@ -22,16 +22,13 @@ const ( controlFramePong uint8 = 5 controlFrameDeviceSnapshot uint8 = 6 controlFrameDeviceDelta uint8 = 7 - controlFrameLeaseRequest uint8 = 8 - controlFrameLeaseResponse uint8 = 9 controlCapabilityChanged uint32 = 1 << 0 controlCapabilityPingPong uint32 = 1 << 1 controlCapabilityPayloadFrames uint32 = 1 << 2 controlCapabilityDeviceStateV2 uint32 = 1 << 3 - controlCapabilityImportLease uint32 = 1 << 4 controlRequiredCapabilities = controlCapabilityChanged | controlCapabilityPingPong - controlExtensionCapabilities = controlCapabilityPayloadFrames | controlCapabilityDeviceStateV2 | controlCapabilityImportLease + controlExtensionCapabilities = controlCapabilityPayloadFrames | controlCapabilityDeviceStateV2 controlCapabilities = controlRequiredCapabilities | controlExtensionCapabilities controlPrefaceSize = 8 @@ -45,10 +42,6 @@ const ( backendIDLinuxSysfs = "linux-sysfs" backendIDDarwinIOKit = "darwin-iokit" backendIDWindowsVBoxUSB = "windows-vboxusb" - - leaseErrorBadRequest = "bad_request" - leaseErrorUnavailable = "unavailable" - leaseErrorBusy = "busy" ) var controlPreface = [controlPrefaceSize]byte{'S', 'B', 'U', 'S', 'B', 'I', 'P', '1'} @@ -106,21 +99,6 @@ type controlDeviceDelta struct { Removed []string `json:"removed,omitempty"` } -type controlLeaseRequest struct { - BusID string `json:"busid"` - ClientNonce uint64 `json:"client_nonce"` -} - -type controlLeaseResponse struct { - BusID string `json:"busid"` - LeaseID uint64 `json:"lease_id,omitempty"` - ClientNonce uint64 `json:"client_nonce"` - Generation uint64 `json:"generation,omitempty"` - TTLMillis int64 `json:"ttl_millis,omitempty"` - ErrorCode string `json:"error_code,omitempty"` - ErrorMessage string `json:"error_message,omitempty"` -} - // controlReader reuses its payload scratch across successive reads on a // single connection. The returned payload is only valid until the next call. type controlReader struct { diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index 188c624a2..2480ed389 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -255,8 +255,6 @@ func (s *darwinFakeUSBIPServer) handleConn(conn net.Conn) { _ = WriteOpRepDevList(conn, []DeviceEntry{s.entry}) case OpReqImport: s.handleImport(conn) - case OpReqImportExt: - s.handleImportExt(conn) } } @@ -292,24 +290,7 @@ func (s *darwinFakeUSBIPServer) handleControlConn(conn net.Conn) { if err != nil { return } - frame := message.Frame - if frame.Type != controlFramePing { - if frame.Type == controlFrameLeaseRequest && supportsControlExtensions(capabilities) { - var request controlLeaseRequest - if unmarshalControlPayload(message.Payload, &request) != nil { - return - } - _ = writeControlMessage(conn, controlFrame{ - Type: controlFrameLeaseResponse, - Version: controlProtocolVersion, - }, controlLeaseResponse{ - BusID: request.BusID, - LeaseID: 1, - ClientNonce: request.ClientNonce, - TTLMillis: int64(importLeaseTTL / time.Millisecond), - }) - continue - } + if message.Frame.Type != controlFramePing { return } err = writeControlMessage(conn, controlFrame{Type: controlFramePong, Version: controlProtocolVersion}, nil) @@ -336,23 +317,6 @@ func (s *darwinFakeUSBIPServer) handleImport(conn net.Conn) { s.handleDataSession(conn) } -func (s *darwinFakeUSBIPServer) handleImportExt(conn net.Conn) { - request, err := ReadOpReqImportExtBody(conn) - if err != nil { - return - } - if request.BusID != s.entry.Info.BusIDString() || request.LeaseID == 0 { - _ = WriteOpRepImport(conn, OpRepImportExt, OpStatusError, nil) - return - } - info := s.entry.Info - err = WriteOpRepImport(conn, OpRepImportExt, OpStatusOK, &info) - if err != nil { - return - } - s.handleDataSession(conn) -} - func (s *darwinFakeUSBIPServer) handleDataSession(conn net.Conn) { for { header, err := ReadDataHeader(conn) diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go index fec10bb41..0712e7b69 100644 --- a/service/usbip/export_ledger.go +++ b/service/usbip/export_ledger.go @@ -18,7 +18,6 @@ import ( type exportLedger struct { logger log.ContextLogger now func() time.Time - ttl time.Duration broadcastAccess sync.Mutex seq uint64 @@ -29,8 +28,6 @@ type exportLedger struct { inventoryAccess sync.Mutex exports map[string]Export busy map[string]bool - leases map[string]serverImportLease - nextLeaseID uint64 } type exportSubscriber struct { @@ -40,34 +37,19 @@ type exportSubscriber struct { send chan controlMessage } -type serverImportLease struct { - ID uint64 - SubscriberID uint64 - BusID string - ClientNonce uint64 - Generation uint64 - Identity ExportLeaseIdentity - Expires time.Time -} +const controlSubscriberSendBuffer = 16 -const ( - controlSubscriberSendBuffer = 16 - importLeaseTTL = 10 * time.Second -) - -func newExportLedger(logger log.ContextLogger, ttl time.Duration, now func() time.Time) *exportLedger { +func newExportLedger(logger log.ContextLogger, now func() time.Time) *exportLedger { if now == nil { now = time.Now } return &exportLedger{ logger: logger, now: now, - ttl: ttl, subs: make(map[uint64]*exportSubscriber), state: make(map[string]DeviceInfoV2), exports: make(map[string]Export), busy: make(map[string]bool), - leases: make(map[string]serverImportLease), } } @@ -99,29 +81,17 @@ func (l *exportLedger) withInventoryWriteQuiet(body func()) { func (l *exportLedger) IsReserved(busid string) bool { var reserved bool l.withInventoryRead(func() { - reserved = l.reservedLocked(busid) + reserved = l.busy[busid] }) return reserved } -// reservedLocked: caller must hold l.inventoryAccess. -func (l *exportLedger) reservedLocked(busid string) bool { - if l.busy[busid] { - return true - } - lease, found := l.leases[busid] - if !found { - return false - } - return l.now().Before(lease.Expires) -} - func (l *exportLedger) AvailableExports() []Export { var out []Export l.withInventoryRead(func() { out = make([]Export, 0, len(l.exports)) for busid, export := range l.exports { - if l.reservedLocked(busid) { + if l.busy[busid] { continue } out = append(out, export) @@ -190,49 +160,32 @@ func (l *exportLedger) BroadcastIfChanged() bool { return true } -// TryReserveForImport runs LeaseCheck outside the lock and re-checks -// availability before marking busy. Caller must pair success with +// TryReserveForImport atomically reserves an exported busid. The single +// critical section under inventoryAccess closes the TOCTOU window between +// availability check and busy mark. Caller must pair success with // ReleaseImport and broadcast once the session is wired up. func (l *exportLedger) TryReserveForImport(busid string) (Export, bool, string) { var ( - export Export - found bool - reserved bool - ) - l.withInventoryRead(func() { - export, found = l.exports[busid] - reserved = found && l.reservedLocked(busid) - }) - if !found { - return nil, false, "unknown busid" - } - identity := export.LeaseIdentity() - if reserved { - return nil, false, deviceStateBusy - } - leaseOK, leaseReason := export.LeaseCheck() - if !leaseOK { - return nil, false, leaseReason - } - var ( - reserveOK bool - failure string + export Export + ok bool + reason string ) l.withInventoryWriteQuiet(func() { - current, stillExported := l.exports[busid] - if !stillExported || current.LeaseIdentity() != identity { - failure = "unknown busid" + current, found := l.exports[busid] + if !found { + reason = "unknown busid" return } - if l.reservedLocked(busid) { - failure = deviceStateBusy + if l.busy[busid] { + reason = deviceStateBusy return } l.busy[busid] = true - reserveOK = true + export = current + ok = true }) - if !reserveOK { - return nil, false, failure + if !ok { + return nil, false, reason } return export, true, "" } @@ -247,231 +200,6 @@ func (l *exportLedger) ReleaseImport(busid string, removeExport bool) { }) } -// IssueLease pins lease correctness to the export identity; the -// broadcast sequence on the response is opaque metadata for clients. -func (l *exportLedger) IssueLease(subID uint64, request controlLeaseRequest) controlLeaseResponse { - response := controlLeaseResponse{ - BusID: request.BusID, - ClientNonce: request.ClientNonce, - } - if request.BusID == "" { - response.ErrorCode = leaseErrorBadRequest - response.ErrorMessage = "missing busid" - return response - } - - l.broadcastAccess.Lock() - generation := l.seq - l.broadcastAccess.Unlock() - - var ( - export Export - identity ExportLeaseIdentity - preCheckOK bool - ) - l.withInventoryWrite(func() bool { - now := l.now() - changed := l.cleanupExpiredLocked(now) - currentExport, found := l.exports[request.BusID] - if !found { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = "unknown busid" - return changed - } - if l.busy[request.BusID] { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = deviceStateBusy - return changed - } - if _, exists := l.leases[request.BusID]; exists { - response.ErrorCode = leaseErrorBusy - response.ErrorMessage = "lease already active" - return changed - } - export = currentExport - identity = currentExport.LeaseIdentity() - preCheckOK = true - return changed - }) - if !preCheckOK { - return response - } - - leaseOK, leaseReason := export.LeaseCheck() - if !leaseOK { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = leaseReason - return response - } - - l.withInventoryWrite(func() bool { - now := l.now() - changed := l.cleanupExpiredLocked(now) - current, stillExported := l.exports[request.BusID] - if !stillExported || current.LeaseIdentity() != identity { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = "unknown busid" - return changed - } - if l.busy[request.BusID] { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = deviceStateBusy - return changed - } - if _, exists := l.leases[request.BusID]; exists { - response.ErrorCode = leaseErrorBusy - response.ErrorMessage = "lease already active" - return changed - } - l.nextLeaseID++ - lease := serverImportLease{ - ID: l.nextLeaseID, - SubscriberID: subID, - BusID: request.BusID, - ClientNonce: request.ClientNonce, - Generation: generation, - Identity: current.LeaseIdentity(), - Expires: now.Add(l.ttl), - } - l.leases[request.BusID] = lease - response.LeaseID = lease.ID - response.Generation = lease.Generation - response.TTLMillis = int64(l.ttl / time.Millisecond) - return true - }) - return response -} - -// ConsumeLeaseAndReserve consumes the lease on every outcome except an -// ID/nonce mismatch (the latter preserves the lease for the legitimate -// holder). Caller must pair success with ReleaseImport. -func (l *exportLedger) ConsumeLeaseAndReserve(request ImportExtRequest) (Export, bool, string) { - var ( - export Export - identity ExportLeaseIdentity - phase1OK bool - reason string - ) - l.withInventoryWrite(func() bool { - now := l.now() - changed := l.cleanupExpiredLocked(now) - - lease, found := l.leases[request.BusID] - if !found { - reason = "lease not found" - return changed - } - if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { - reason = "lease mismatch" - return changed - } - if !now.Before(lease.Expires) { - delete(l.leases, request.BusID) - reason = "lease expired" - return true - } - current, stillExported := l.exports[request.BusID] - if !stillExported { - delete(l.leases, request.BusID) - reason = "unknown busid" - return true - } - identity = current.LeaseIdentity() - if identity != lease.Identity { - delete(l.leases, request.BusID) - reason = "lease stale" - return true - } - if l.busy[request.BusID] { - delete(l.leases, request.BusID) - reason = deviceStateBusy - return true - } - export = current - phase1OK = true - return changed - }) - if !phase1OK { - return nil, false, reason - } - - leaseOK, leaseReason := export.LeaseCheck() - if !leaseOK { - l.withInventoryWrite(func() bool { - currentLease, exists := l.leases[request.BusID] - if !exists { - return false - } - if currentLease.ID != request.LeaseID || currentLease.ClientNonce != request.ClientNonce { - return false - } - delete(l.leases, request.BusID) - return true - }) - return nil, false, leaseReason - } - - var ( - finalExport Export - finalOK bool - ) - l.withInventoryWrite(func() bool { - now := l.now() - changed := l.cleanupExpiredLocked(now) - - lease, found := l.leases[request.BusID] - if !found { - reason = "lease not found" - return changed - } - if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { - reason = "lease mismatch" - return changed - } - if !now.Before(lease.Expires) { - delete(l.leases, request.BusID) - reason = "lease expired" - return true - } - current, stillExported := l.exports[request.BusID] - if !stillExported { - delete(l.leases, request.BusID) - reason = "unknown busid" - return true - } - if lease.Identity != identity || current.LeaseIdentity() != identity { - delete(l.leases, request.BusID) - reason = "lease stale" - return true - } - if l.busy[request.BusID] { - delete(l.leases, request.BusID) - reason = deviceStateBusy - return true - } - delete(l.leases, request.BusID) - l.busy[request.BusID] = true - finalExport = current - finalOK = true - return true - }) - if !finalOK { - return nil, false, reason - } - return finalExport, true, "" -} - -func (l *exportLedger) cleanupExpiredLocked(now time.Time) bool { - changed := false - for busid, lease := range l.leases { - if !now.Before(lease.Expires) { - delete(l.leases, busid) - changed = true - } - } - return changed -} - // Subscribe enqueues a freshly computed snapshot to extension-capable // subscribers so they see current state regardless of when the last // broadcast fired. Does NOT mutate l.state: other subscribers must @@ -524,24 +252,11 @@ func (l *exportLedger) Subscribe(conn net.Conn, capabilities uint32) (*exportSub } // Unsubscribe leaves the subscriber's send channel for the GC to -// reclaim; the transport read loop has already exited. Any leases the -// subscriber held are released and the resulting state change is -// broadcast so remaining subscribers see the busid become available -// again. +// reclaim; the transport read loop has already exited. func (l *exportLedger) Unsubscribe(sub *exportSubscriber) { l.broadcastAccess.Lock() delete(l.subs, sub.id) l.broadcastAccess.Unlock() - l.withInventoryWrite(func() bool { - released := false - for busid, lease := range l.leases { - if lease.SubscriberID == sub.id { - delete(l.leases, busid) - released = true - } - } - return released - }) } // CloseAllSubscribers returns the underlying connections so the caller @@ -561,36 +276,9 @@ func (l *exportLedger) ResetForClose() { l.withInventoryWriteQuiet(func() { l.exports = make(map[string]Export) l.busy = make(map[string]bool) - l.leases = make(map[string]serverImportLease) }) } -func (l *exportLedger) HandleControlLeaseRequest(sub *exportSubscriber, payload []byte) { - var request controlLeaseRequest - err := unmarshalControlPayload(payload, &request) - if err != nil { - l.broadcastAccess.Lock() - sequence := l.seq - l.broadcastAccess.Unlock() - l.enqueuePayload(sub, controlFrame{ - Type: controlFrameLeaseResponse, - Version: controlProtocolVersion, - }, controlLeaseResponse{ - ErrorCode: leaseErrorBadRequest, - ErrorMessage: err.Error(), - }, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: sequence}) - return - } - response := l.IssueLease(sub.id, request) - l.broadcastAccess.Lock() - sequence := l.seq - l.broadcastAccess.Unlock() - l.enqueuePayload(sub, controlFrame{ - Type: controlFrameLeaseResponse, - Version: controlProtocolVersion, - }, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: sequence}) -} - func (l *exportLedger) snapshotDeviceState() []DeviceInfoV2 { type entry struct { export Export @@ -600,7 +288,7 @@ func (l *exportLedger) snapshotDeviceState() []DeviceInfoV2 { l.withInventoryRead(func() { entries = make([]entry, 0, len(l.exports)) for busid, export := range l.exports { - entries = append(entries, entry{export: export, busy: l.reservedLocked(busid)}) + entries = append(entries, entry{export: export, busy: l.busy[busid]}) } }) if len(entries) == 0 { diff --git a/service/usbip/export_ledger_darwin_test.go b/service/usbip/export_ledger_darwin_test.go index 68dfeff04..423a37e02 100644 --- a/service/usbip/export_ledger_darwin_test.go +++ b/service/usbip/export_ledger_darwin_test.go @@ -12,7 +12,7 @@ import ( func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) { t.Parallel() - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) + ledger := newExportLedger(nil, func() time.Time { return time.Unix(0, 0) }) entry := darwinFakeDeviceEntry() export := &darwinExport{ busid: entry.Info.BusIDString(), diff --git a/service/usbip/export_ledger_test.go b/service/usbip/export_ledger_test.go index 75208b988..a210885aa 100644 --- a/service/usbip/export_ledger_test.go +++ b/service/usbip/export_ledger_test.go @@ -10,7 +10,7 @@ import ( ) func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) { - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) + ledger := newExportLedger(nil, func() time.Time { return time.Unix(0, 0) }) oldExport := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001} newExport := &testExport{busid: "2-1", vendorID: 0x2222, productID: 0x0002} @@ -53,156 +53,11 @@ func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) { } } -func TestConsumeLeaseAndReserveRejectsIdentityReplacement(t *testing.T) { - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) - original := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001, identity: "linux:original"} - replacement := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001, identity: "linux:replacement"} - - ledger.ApplyHostSnapshot(map[string]Export{original.busid: original}, nil) - lease := ledger.IssueLease(1, controlLeaseRequest{BusID: original.busid, ClientNonce: 7}) - ledger.ApplyHostSnapshot(map[string]Export{replacement.busid: replacement}, nil) - - _, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{ - BusID: original.busid, - LeaseID: lease.LeaseID, - ClientNonce: lease.ClientNonce, - }) - if ok { - t.Fatal("expected replaced export lease to be rejected") - } - if reason != "lease stale" { - t.Fatalf("expected lease stale, got %q", reason) - } -} - -func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) { - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) - available := true - exp := &testExport{ - busid: "1-1", - vendorID: 0x1111, - productID: 0x0001, - leaseCheck: func() (bool, string) { - if available { - return true, "" - } - return false, "capture released" - }, - } - - ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil) - lease := ledger.IssueLease(1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 9}) - available = false - - _, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{ - BusID: exp.busid, - LeaseID: lease.LeaseID, - ClientNonce: lease.ClientNonce, - }) - if ok { - t.Fatal("expected unavailable export lease to be rejected") - } - if reason != "capture released" { - t.Fatalf("expected capture released, got %q", reason) - } - - _, ok, reason = ledger.ConsumeLeaseAndReserve(ImportExtRequest{ - BusID: exp.busid, - LeaseID: lease.LeaseID, - ClientNonce: lease.ClientNonce, - }) - if ok { - t.Fatal("expected consumed lease to stay unavailable on retry") - } - if reason != "lease not found" { - t.Fatalf("expected consumed lease to disappear, got %q", reason) - } -} - -func TestUnsubscribeBroadcastsLeaseRelease(t *testing.T) { - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) - exp := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001} - - ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil) - ledger.SeedBroadcastState() - - holder, _ := ledger.Subscribe(nil, controlCapabilities) - drainSubscriber(holder) - - response := ledger.IssueLease(holder.id, controlLeaseRequest{BusID: exp.busid, ClientNonce: 7}) - if response.ErrorCode != "" { - t.Fatalf("IssueLease failed: %s", response.ErrorMessage) - } - // Pretend a topology event flushed the busy state into the broadcast - // baseline, matching the real-world scenario where hotplug churn keeps - // l.state roughly in sync with reality. Without this the diff machinery - // in BroadcastIfChanged has no busy→available transition to emit. - ledger.BroadcastIfChanged() - drainSubscriber(holder) - - observer, _ := ledger.Subscribe(nil, controlCapabilities) - drainSubscriber(observer) - - ledger.Unsubscribe(holder) - - select { - case msg := <-observer.send: - if msg.Frame.Type != controlFrameDeviceDelta { - t.Fatalf("expected device delta after lease release, got frame type %d", msg.Frame.Type) - } - var delta controlDeviceDelta - err := unmarshalControlPayload(msg.Payload, &delta) - if err != nil { - t.Fatal(err) - } - if len(delta.Updated) != 1 || delta.Updated[0].State != deviceStateAvailable { - t.Fatalf("expected one Updated entry flipped to available, got %#v", delta) - } - default: - t.Fatal("expected observer to receive a delta after holder's lease was released") - } -} - -func drainSubscriber(sub *exportSubscriber) { - for { - select { - case <-sub.send: - default: - return - } - } -} - -func TestConsumeLeaseAndReserveMarksBusyOnSuccess(t *testing.T) { - ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) - exp := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001} - - ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil) - lease := ledger.IssueLease(1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 11}) - - reserved, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{ - BusID: exp.busid, - LeaseID: lease.LeaseID, - ClientNonce: lease.ClientNonce, - }) - if !ok { - t.Fatalf("expected lease reservation success, got %q", reason) - } - if reserved != exp { - t.Fatal("expected to reserve the original export instance") - } - if !ledger.IsReserved(exp.busid) { - t.Fatal("expected successful lease reservation to mark busid reserved") - } -} - type testExport struct { busid string vendorID uint16 productID uint16 - identity ExportLeaseIdentity - leaseCheck func() (bool, string) onSnapshot func() } @@ -210,13 +65,6 @@ func (e *testExport) BusID() string { return e.busid } -func (e *testExport) LeaseIdentity() ExportLeaseIdentity { - if e.identity != "" { - return e.identity - } - return ExportLeaseIdentity(e.busid) -} - func (e *testExport) Snapshot(busy bool) ExportSnapshot { onSnapshot := e.onSnapshot e.onSnapshot = nil @@ -235,13 +83,6 @@ func (e *testExport) Snapshot(busy bool) ExportSnapshot { } } -func (e *testExport) LeaseCheck() (bool, string) { - if e.leaseCheck != nil { - return e.leaseCheck() - } - return true, "" -} - func (e *testExport) DeviceInfo() (DeviceInfoTruncated, error) { return e.deviceInfo(), nil } diff --git a/service/usbip/host.go b/service/usbip/host.go index 40df0eb76..f57447a23 100644 --- a/service/usbip/host.go +++ b/service/usbip/host.go @@ -26,8 +26,6 @@ type ImportHost interface { Attach(ctx context.Context, info DeviceInfoTruncated, conn net.Conn) (AttachedSession, error) } -type ExportLeaseIdentity string - // Export pointers handed back from Reconcile are immutable from the // ledger's perspective: the ledger calls the methods below outside any // lock. Hosts that need to mutate must clone, mutate the clone, then @@ -35,8 +33,6 @@ type ExportLeaseIdentity string type Export interface { BusID() string Snapshot(busy bool) ExportSnapshot - LeaseIdentity() ExportLeaseIdentity - LeaseCheck() (ok bool, reason string) DeviceInfo() (DeviceInfoTruncated, error) NewServerDataSession(ctx context.Context, conn net.Conn) (DataSession, error) } diff --git a/service/usbip/host_darwin.go b/service/usbip/host_darwin.go index d7385c281..a3ace6def 100644 --- a/service/usbip/host_darwin.go +++ b/service/usbip/host_darwin.go @@ -294,10 +294,6 @@ func (e *darwinExport) BusID() string { return e.busid } -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" @@ -328,13 +324,6 @@ func (e *darwinExport) Snapshot(busy bool) ExportSnapshot { } } -func (e *darwinExport) LeaseCheck() (bool, string) { - if e.stale { - return false, e.staleReason() - } - return true, "" -} - func (e *darwinExport) DeviceInfo() (DeviceInfoTruncated, error) { return e.entry.Info, nil } diff --git a/service/usbip/host_linux.go b/service/usbip/host_linux.go index ec9e6bade..b9d7e915c 100644 --- a/service/usbip/host_linux.go +++ b/service/usbip/host_linux.go @@ -112,29 +112,6 @@ func (i linuxExportIdentity) Equal(other linuxExportIdentity) bool { return true } -func (i linuxExportIdentity) LeaseIdentity() ExportLeaseIdentity { - var builder strings.Builder - fmt.Fprintf(&builder, "linux:%d:%d:%d:%04x:%04x:%04x:%02x:%02x:%02x:%02x:%02x:%02x:%s", - i.BusNum, - i.DevNum, - i.Speed, - i.VendorID, - i.ProductID, - i.BCDDevice, - i.DeviceClass, - i.DeviceSubClass, - i.DeviceProtocol, - i.ConfigValue, - i.NumConfigs, - i.NumInterfaces, - i.Serial, - ) - for _, iface := range i.Interfaces { - fmt.Fprintf(&builder, "|%02x.%02x.%02x", iface.BInterfaceClass, iface.BInterfaceSubClass, iface.BInterfaceProtocol) - } - return ExportLeaseIdentity(builder.String()) -} - type linuxExportHost struct { logger log.ContextLogger matches []option.USBIPDeviceMatch @@ -658,10 +635,6 @@ func (e *linuxExport) BusID() string { return e.busid } -func (e *linuxExport) LeaseIdentity() ExportLeaseIdentity { - return e.identity.LeaseIdentity() -} - func (e *linuxExport) Snapshot(busy bool) ExportSnapshot { stableID := "linux-busid:" + e.descriptor.BusID if e.descriptor.Serial != "" { @@ -714,20 +687,6 @@ func (e *linuxExport) Snapshot(busy bool) ExportSnapshot { } } -func (e *linuxExport) LeaseCheck() (bool, string) { - if e.stale { - return false, "device replaced" - } - status, err := readUsbipStatus(e.busid) - if err != nil { - return false, err.Error() - } - if status != usbipStatusAvailable { - return false, linuxUSBIPStatusReason(status) - } - return true, "" -} - func (e *linuxExport) DeviceInfo() (DeviceInfoTruncated, error) { return e.descriptor.toProtocol(), nil } diff --git a/service/usbip/host_windows.go b/service/usbip/host_windows.go index 768656a02..af23289db 100644 --- a/service/usbip/host_windows.go +++ b/service/usbip/host_windows.go @@ -296,14 +296,6 @@ func (e *windowsExport) BusID() string { return e.info.BusID } -func (e *windowsExport) LeaseIdentity() ExportLeaseIdentity { - return ExportLeaseIdentity("windows:" + e.info.InstanceID) -} - -func (e *windowsExport) LeaseCheck() (bool, string) { - return true, "" -} - func (e *windowsExport) Snapshot(busy bool) ExportSnapshot { state := deviceStateAvailable if busy { diff --git a/service/usbip/protocol.go b/service/usbip/protocol.go index 62fa6f895..8f00901f9 100644 --- a/service/usbip/protocol.go +++ b/service/usbip/protocol.go @@ -15,12 +15,10 @@ const ( ProtocolVersion uint16 = 0x0111 - OpReqDevList uint16 = 0x8005 - OpRepDevList uint16 = 0x0005 - OpReqImport uint16 = 0x8003 - OpRepImport uint16 = 0x0003 - OpReqImportExt uint16 = 0x8f03 - OpRepImportExt uint16 = 0x0f03 + OpReqDevList uint16 = 0x8005 + OpRepDevList uint16 = 0x0005 + OpReqImport uint16 = 0x8003 + OpRepImport uint16 = 0x0003 OpStatusOK uint32 = 0 OpStatusError uint32 = 1 @@ -29,7 +27,6 @@ const ( maxOpRepDevListBodyBytes = 8 << 20 deviceInfoWireSize = 312 deviceInterfaceWireSize = 4 - importExtBodyWireSize = 56 ) const ( @@ -78,13 +75,6 @@ type DeviceEntry struct { Serial string } -type ImportExtRequest struct { - BusID string - LeaseID uint64 - ClientNonce uint64 - Flags uint32 -} - func WriteOpHeader(w io.Writer, code uint16, status uint32) error { return binary.Write(w, binary.BigEndian, OpHeader{ Version: ProtocolVersion, @@ -123,23 +113,6 @@ func WriteOpReqImport(w io.Writer, busid string) error { return binary.Write(w, binary.BigEndian, field) } -func WriteOpReqImportExt(w io.Writer, request ImportExtRequest) error { - err := WriteOpHeader(w, OpReqImportExt, OpStatusOK) - if err != nil { - return err - } - var raw [importExtBodyWireSize]byte - if len(request.BusID) >= 32 { - return E.New("busid too long: ", request.BusID) - } - copy(raw[:32], request.BusID) - binary.BigEndian.PutUint64(raw[32:40], request.LeaseID) - binary.BigEndian.PutUint64(raw[40:48], request.ClientNonce) - binary.BigEndian.PutUint32(raw[48:52], request.Flags) - _, err = w.Write(raw[:]) - return err -} - func ReadOpReqImportBody(r io.Reader) (string, error) { var field [32]byte _, err := io.ReadFull(r, field[:]) @@ -149,20 +122,6 @@ func ReadOpReqImportBody(r io.Reader) (string, error) { return cstring(field[:]), nil } -func ReadOpReqImportExtBody(r io.Reader) (ImportExtRequest, error) { - var raw [importExtBodyWireSize]byte - _, err := io.ReadFull(r, raw[:]) - if err != nil { - return ImportExtRequest{}, err - } - return ImportExtRequest{ - BusID: cstring(raw[:32]), - LeaseID: binary.BigEndian.Uint64(raw[32:40]), - ClientNonce: binary.BigEndian.Uint64(raw[40:48]), - Flags: binary.BigEndian.Uint32(raw[48:52]), - }, nil -} - func WriteOpRepImport(w io.Writer, code uint16, status uint32, info *DeviceInfoTruncated) error { err := WriteOpHeader(w, code, status) if err != nil { diff --git a/service/usbip/server.go b/service/usbip/server.go index aa45954e4..241656795 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -62,7 +62,7 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string, logger: logger, matches: options.Devices, host: host, - ledger: newExportLedger(logger, importLeaseTTL, time.Now), + ledger: newExportLedger(logger, time.Now), sessions: make(map[DataSession]struct{}), listener: listener.New(listener.Options{ Context: ctx, @@ -200,8 +200,6 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) { break } closeConn = !s.handleImportBusID(conn, busid) - case OpReqImportExt: - closeConn = !s.handleImportExt(conn) default: s.logger.Debug(fmt.Sprintf("unknown opcode 0x%04x", header.Code)) } @@ -275,21 +273,6 @@ func (s *ServerService) buildDevListEntries() []DeviceEntry { return entries } -func (s *ServerService) handleImportExt(conn net.Conn) bool { - request, err := ReadOpReqImportExtBody(conn) - if err != nil { - s.logger.Debug("read import-ext body: ", err) - return false - } - export, ok, reason := s.ledger.ConsumeLeaseAndReserve(request) - if !ok { - s.logger.Info("import-ext rejected (", request.BusID, ": ", reason, ")") - _ = WriteOpRepImport(conn, OpRepImportExt, OpStatusError, nil) - return false - } - return s.handleImportReserved(conn, request.BusID, export, true) -} - func (s *ServerService) handleImportBusID(conn net.Conn, busid string) bool { export, ok, reason := s.ledger.TryReserveForImport(busid) if !ok { @@ -297,30 +280,26 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string) bool { _ = WriteOpRepImport(conn, OpRepImport, OpStatusError, nil) return false } - return s.handleImportReserved(conn, busid, export, false) + return s.handleImportReserved(conn, busid, export) } -func (s *ServerService) handleImportReserved(conn net.Conn, busid string, export Export, extended bool) bool { - opCode := uint16(OpRepImport) - if extended { - opCode = OpRepImportExt - } +func (s *ServerService) handleImportReserved(conn net.Conn, busid string, export Export) bool { info, err := export.DeviceInfo() if err != nil { s.ledger.ReleaseImport(busid, false) s.logger.Warn("refresh ", busid, ": ", err) - _ = WriteOpRepImport(conn, opCode, OpStatusError, nil) + _ = WriteOpRepImport(conn, OpRepImport, OpStatusError, nil) return false } session, err := export.NewServerDataSession(s.ctx, conn) if err != nil { s.ledger.ReleaseImport(busid, false) s.logger.Warn("open data session ", busid, ": ", err) - _ = WriteOpRepImport(conn, opCode, OpStatusError, nil) + _ = WriteOpRepImport(conn, OpRepImport, OpStatusError, nil) return false } s.ledger.BroadcastIfChanged() - err = WriteOpRepImport(conn, opCode, OpStatusOK, &info) + err = WriteOpRepImport(conn, OpRepImport, OpStatusOK, &info) if err != nil { s.logger.Warn("reply import ", busid, ": ", err) s.tearDownPreparedSession(busid, session) diff --git a/service/usbip/server_shared.go b/service/usbip/server_shared.go index 99bd2aaaf..2d55e1c60 100644 --- a/service/usbip/server_shared.go +++ b/service/usbip/server_shared.go @@ -70,12 +70,6 @@ func (s *ServerService) readControlConn(sub *exportSubscriber, done chan<- struc Type: controlFramePong, Version: controlProtocolVersion, }) - case controlFrameLeaseRequest: - if supportsControlExtensions(sub.capabilities) { - s.ledger.HandleControlLeaseRequest(sub, message.Payload) - continue - } - return default: return }