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.
This commit is contained in:
世界
+3
-23
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
+21
-333
@@ -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 {
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+6
-27
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user