usbip: add extended control protocol
This commit is contained in:
世界
@@ -0,0 +1,208 @@
|
||||
//go:build linux || (darwin && cgo)
|
||||
|
||||
package usbip
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
E "github.com/sagernet/sing/common/exceptions"
|
||||
)
|
||||
|
||||
var errClientControlSessionClosed = errors.New("usbip control session closed")
|
||||
|
||||
type clientImportLease struct {
|
||||
Valid bool
|
||||
ID uint64
|
||||
ClientNonce uint64
|
||||
}
|
||||
|
||||
type clientControlSession struct {
|
||||
conn net.Conn
|
||||
capabilities uint32
|
||||
writeMu sync.Mutex
|
||||
mu 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) supportsImportLease() bool {
|
||||
return supportsControlExtensions(s.capabilities)
|
||||
}
|
||||
|
||||
func (s *clientControlSession) writeControl(frame controlFrame, payload any) error {
|
||||
s.writeMu.Lock()
|
||||
defer s.writeMu.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.mu.Lock()
|
||||
if s.closed {
|
||||
s.mu.Unlock()
|
||||
return controlLeaseResponse{}, errClientControlSessionClosed
|
||||
}
|
||||
s.nextNonce++
|
||||
nonce := s.nextNonce
|
||||
waiter := make(chan clientLeaseResult, 1)
|
||||
s.pending[nonce] = waiter
|
||||
s.mu.Unlock()
|
||||
|
||||
request := controlLeaseRequest{
|
||||
BusID: busid,
|
||||
ClientNonce: nonce,
|
||||
}
|
||||
if err := s.writeControl(controlFrame{
|
||||
Type: controlFrameLeaseRequest,
|
||||
Version: controlProtocolVersion,
|
||||
}, request); err != nil {
|
||||
s.removeLeaseWaiter(nonce)
|
||||
return controlLeaseResponse{}, err
|
||||
}
|
||||
|
||||
select {
|
||||
case result := <-waiter:
|
||||
return result.response, result.err
|
||||
case <-ctx.Done():
|
||||
s.removeLeaseWaiter(nonce)
|
||||
return controlLeaseResponse{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *clientControlSession) deliverLeaseResponse(response controlLeaseResponse) bool {
|
||||
s.mu.Lock()
|
||||
waiter, ok := s.pending[response.ClientNonce]
|
||||
if ok {
|
||||
delete(s.pending, response.ClientNonce)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
waiter <- clientLeaseResult{response: response}
|
||||
close(waiter)
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *clientControlSession) removeLeaseWaiter(nonce uint64) {
|
||||
s.mu.Lock()
|
||||
delete(s.pending, nonce)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *clientControlSession) closeWithError(err error) {
|
||||
s.mu.Lock()
|
||||
if s.closed {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.closed = true
|
||||
pending := s.pending
|
||||
s.pending = make(map[uint64]chan clientLeaseResult)
|
||||
s.mu.Unlock()
|
||||
|
||||
for _, waiter := range pending {
|
||||
waiter <- clientLeaseResult{err: err}
|
||||
close(waiter)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ClientService) setControlSession(session *clientControlSession) {
|
||||
c.controlMu.Lock()
|
||||
c.controlSession = session
|
||||
c.controlMu.Unlock()
|
||||
}
|
||||
|
||||
func (c *ClientService) clearControlSession(session *clientControlSession, err error) {
|
||||
c.controlMu.Lock()
|
||||
if c.controlSession == session {
|
||||
c.controlSession = nil
|
||||
}
|
||||
c.controlMu.Unlock()
|
||||
session.closeWithError(err)
|
||||
}
|
||||
|
||||
func (c *ClientService) currentControlSession() *clientControlSession {
|
||||
c.controlMu.Lock()
|
||||
defer c.controlMu.Unlock()
|
||||
return c.controlSession
|
||||
}
|
||||
|
||||
func (c *ClientService) requestImportLease(ctx context.Context, busid string) (clientImportLease, error) {
|
||||
session := c.currentControlSession()
|
||||
if session == nil || !session.supportsImportLease() {
|
||||
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) applyControlSnapshot(snapshot controlDeviceSnapshot) {
|
||||
devices := deviceInfoV2Map(snapshot.Devices)
|
||||
values := sortedDeviceInfoV2Values(devices)
|
||||
c.remoteMu.Lock()
|
||||
c.remoteDevicesV2 = devices
|
||||
c.remoteMu.Unlock()
|
||||
c.applyRemoteEntries(deviceInfoV2ToEntries(values, true))
|
||||
}
|
||||
|
||||
func (c *ClientService) applyControlDelta(delta controlDeviceDelta) {
|
||||
c.remoteMu.Lock()
|
||||
if c.remoteDevicesV2 == nil {
|
||||
c.remoteDevicesV2 = make(map[string]DeviceInfoV2)
|
||||
}
|
||||
for _, busid := range delta.Removed {
|
||||
delete(c.remoteDevicesV2, busid)
|
||||
}
|
||||
for _, device := range delta.Added {
|
||||
if device.BusID == "" {
|
||||
continue
|
||||
}
|
||||
c.remoteDevicesV2[device.BusID] = device
|
||||
}
|
||||
for _, device := range delta.Updated {
|
||||
if device.BusID == "" {
|
||||
continue
|
||||
}
|
||||
c.remoteDevicesV2[device.BusID] = device
|
||||
}
|
||||
values := sortedDeviceInfoV2Values(c.remoteDevicesV2)
|
||||
c.remoteMu.Unlock()
|
||||
c.applyRemoteEntries(deviceInfoV2ToEntries(values, true))
|
||||
}
|
||||
|
||||
func (c *ClientService) clearControlDeviceState() {
|
||||
c.remoteMu.Lock()
|
||||
c.remoteDevicesV2 = nil
|
||||
c.remoteMu.Unlock()
|
||||
}
|
||||
@@ -68,6 +68,12 @@ type ClientService struct {
|
||||
|
||||
activeMu sync.Mutex
|
||||
activeBusIDs map[string]struct{}
|
||||
|
||||
controlMu sync.Mutex
|
||||
controlSession *clientControlSession
|
||||
|
||||
remoteMu sync.Mutex
|
||||
remoteDevicesV2 map[string]DeviceInfoV2
|
||||
}
|
||||
|
||||
func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) {
|
||||
@@ -198,18 +204,23 @@ func (c *ClientService) runControlSession() error {
|
||||
if err != nil {
|
||||
return E.Cause(errControlUnsupported, "read control ack: ", err)
|
||||
}
|
||||
if ack.Type != controlFrameAck || ack.Version != controlProtocolVersion || ack.Capabilities&controlCapabilities != controlCapabilities {
|
||||
if ack.Type != controlFrameAck || ack.Version != controlProtocolVersion || ack.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities {
|
||||
return E.Cause(errControlUnsupported, "invalid control ack")
|
||||
}
|
||||
_ = conn.SetWriteDeadline(time.Time{})
|
||||
_ = conn.SetReadDeadline(time.Time{})
|
||||
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
session := newClientControlSession(conn, ack.Capabilities)
|
||||
extended := supportsControlExtensions(ack.Capabilities)
|
||||
if extended {
|
||||
c.setControlSession(session)
|
||||
defer c.clearControlSession(session, errClientControlSessionClosed)
|
||||
} else if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(err, "initial devlist sync")
|
||||
}
|
||||
|
||||
pingDone := make(chan struct{})
|
||||
go c.controlPingLoop(conn, pingDone)
|
||||
go c.controlPingLoop(session, pingDone)
|
||||
defer close(pingDone)
|
||||
|
||||
lastSeq := ack.Sequence
|
||||
@@ -217,19 +228,57 @@ func (c *ClientService) runControlSession() error {
|
||||
if err := conn.SetReadDeadline(time.Now().Add(controlReadTimeout)); err != nil {
|
||||
return err
|
||||
}
|
||||
frame, err := ReadControlFrame(conn)
|
||||
message, err := readControlMessage(conn)
|
||||
if err != nil {
|
||||
return E.Cause(errImmediateReconnect, controlSessionIdleHint, ": ", err)
|
||||
}
|
||||
frame := message.Frame
|
||||
switch frame.Type {
|
||||
case controlFrameChanged:
|
||||
if frame.Sequence != lastSeq+1 {
|
||||
if frame.Sequence != lastSeq && frame.Sequence != lastSeq+1 {
|
||||
return E.Cause(errImmediateReconnect, "control sequence jumped from ", lastSeq, " to ", frame.Sequence)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "devlist sync after change ", frame.Sequence, ": ", err)
|
||||
}
|
||||
case controlFrameDeviceSnapshot:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
var snapshot controlDeviceSnapshot
|
||||
if err := unmarshalControlPayload(message.Payload, &snapshot); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "read device snapshot: ", err)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
c.applyControlSnapshot(snapshot)
|
||||
case controlFrameDeviceDelta:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
if frame.Sequence != lastSeq+1 {
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "devlist sync after sequence jump ", frame.Sequence, ": ", err)
|
||||
}
|
||||
c.clearControlDeviceState()
|
||||
lastSeq = frame.Sequence
|
||||
continue
|
||||
}
|
||||
var delta controlDeviceDelta
|
||||
if err := unmarshalControlPayload(message.Payload, &delta); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "read device delta: ", err)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
c.applyControlDelta(delta)
|
||||
case controlFrameLeaseResponse:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
var response controlLeaseResponse
|
||||
if err := unmarshalControlPayload(message.Payload, &response); 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)
|
||||
@@ -245,7 +294,7 @@ func (c *ClientService) runStandardSession() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) {
|
||||
func (c *ClientService) controlPingLoop(session *clientControlSession, done <-chan struct{}) {
|
||||
ticker := time.NewTicker(controlPingInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
@@ -255,12 +304,13 @@ func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) {
|
||||
case <-done:
|
||||
return
|
||||
case <-ticker.C:
|
||||
_ = conn.SetWriteDeadline(time.Now().Add(controlWriteTimeout))
|
||||
if err := WriteControlPing(conn); err != nil {
|
||||
_ = conn.Close()
|
||||
if err := session.writeControl(controlFrame{
|
||||
Type: controlFramePing,
|
||||
Version: controlProtocolVersion,
|
||||
}, nil); err != nil {
|
||||
_ = session.conn.Close()
|
||||
return
|
||||
}
|
||||
_ = conn.SetWriteDeadline(time.Time{})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -520,7 +570,21 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (*darwi
|
||||
}()
|
||||
stopCloseOnCancel := closeConnOnContextDone(ctx, conn)
|
||||
defer stopCloseOnCancel()
|
||||
if err := WriteOpReqImport(conn, busid); err != nil {
|
||||
lease, err := c.requestImportLease(ctx, busid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
expectedReply := OpRepImport
|
||||
if lease.Valid {
|
||||
expectedReply = OpRepImportExt
|
||||
if err := WriteOpReqImportExt(conn, ImportExtRequest{
|
||||
BusID: busid,
|
||||
LeaseID: lease.ID,
|
||||
ClientNonce: lease.ClientNonce,
|
||||
}); err != nil {
|
||||
return nil, E.Cause(err, "write OP_REQ_IMPORT_EXT")
|
||||
}
|
||||
} else if err := WriteOpReqImport(conn, busid); err != nil {
|
||||
return nil, E.Cause(err, "write OP_REQ_IMPORT")
|
||||
}
|
||||
header, err := ReadOpHeader(conn)
|
||||
@@ -530,7 +594,7 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (*darwi
|
||||
if header.Version != ProtocolVersion {
|
||||
return nil, E.New("unexpected reply version 0x", hex16(header.Version))
|
||||
}
|
||||
if header.Code != OpRepImport {
|
||||
if header.Code != expectedReply {
|
||||
return nil, E.New("unexpected reply code 0x", hex16(header.Code))
|
||||
}
|
||||
if header.Status != OpStatusOK {
|
||||
|
||||
@@ -80,6 +80,12 @@ type ClientService struct {
|
||||
|
||||
activeMu sync.Mutex
|
||||
activeBusIDs map[string]struct{}
|
||||
|
||||
controlMu sync.Mutex
|
||||
controlSession *clientControlSession
|
||||
|
||||
remoteMu sync.Mutex
|
||||
remoteDevicesV2 map[string]DeviceInfoV2
|
||||
}
|
||||
|
||||
func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) {
|
||||
@@ -221,18 +227,23 @@ func (c *ClientService) runControlSession() error {
|
||||
if ack.Version != controlProtocolVersion {
|
||||
return E.Cause(errControlUnsupported, "unsupported control version ", ack.Version)
|
||||
}
|
||||
if ack.Capabilities&controlCapabilities != controlCapabilities {
|
||||
if ack.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities {
|
||||
return E.Cause(errControlUnsupported, "missing control capabilities 0x", ack.Capabilities)
|
||||
}
|
||||
_ = conn.SetWriteDeadline(time.Time{})
|
||||
_ = conn.SetReadDeadline(time.Time{})
|
||||
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
session := newClientControlSession(conn, ack.Capabilities)
|
||||
extended := supportsControlExtensions(ack.Capabilities)
|
||||
if extended {
|
||||
c.setControlSession(session)
|
||||
defer c.clearControlSession(session, errClientControlSessionClosed)
|
||||
} else if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(err, "initial devlist sync")
|
||||
}
|
||||
|
||||
pingDone := make(chan struct{})
|
||||
go c.controlPingLoop(conn, pingDone)
|
||||
go c.controlPingLoop(session, pingDone)
|
||||
defer close(pingDone)
|
||||
|
||||
lastSeq := ack.Sequence
|
||||
@@ -240,19 +251,57 @@ func (c *ClientService) runControlSession() error {
|
||||
if err := conn.SetReadDeadline(time.Now().Add(controlReadTimeout)); err != nil {
|
||||
return err
|
||||
}
|
||||
frame, err := ReadControlFrame(conn)
|
||||
message, err := readControlMessage(conn)
|
||||
if err != nil {
|
||||
return E.Cause(errImmediateReconnect, controlSessionIdleHint, ": ", err)
|
||||
}
|
||||
frame := message.Frame
|
||||
switch frame.Type {
|
||||
case controlFrameChanged:
|
||||
if frame.Sequence != lastSeq+1 {
|
||||
if frame.Sequence != lastSeq && frame.Sequence != lastSeq+1 {
|
||||
return E.Cause(errImmediateReconnect, "control sequence jumped from ", lastSeq, " to ", frame.Sequence)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "devlist sync after change ", frame.Sequence, ": ", err)
|
||||
}
|
||||
case controlFrameDeviceSnapshot:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
var snapshot controlDeviceSnapshot
|
||||
if err := unmarshalControlPayload(message.Payload, &snapshot); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "read device snapshot: ", err)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
c.applyControlSnapshot(snapshot)
|
||||
case controlFrameDeviceDelta:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
if frame.Sequence != lastSeq+1 {
|
||||
if err := c.syncRemoteState(); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "devlist sync after sequence jump ", frame.Sequence, ": ", err)
|
||||
}
|
||||
c.clearControlDeviceState()
|
||||
lastSeq = frame.Sequence
|
||||
continue
|
||||
}
|
||||
var delta controlDeviceDelta
|
||||
if err := unmarshalControlPayload(message.Payload, &delta); err != nil {
|
||||
return E.Cause(errImmediateReconnect, "read device delta: ", err)
|
||||
}
|
||||
lastSeq = frame.Sequence
|
||||
c.applyControlDelta(delta)
|
||||
case controlFrameLeaseResponse:
|
||||
if !extended {
|
||||
return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type)
|
||||
}
|
||||
var response controlLeaseResponse
|
||||
if err := unmarshalControlPayload(message.Payload, &response); 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)
|
||||
@@ -268,7 +317,7 @@ func (c *ClientService) runStandardSession() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) {
|
||||
func (c *ClientService) controlPingLoop(session *clientControlSession, done <-chan struct{}) {
|
||||
ticker := time.NewTicker(controlPingInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
@@ -278,12 +327,13 @@ func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) {
|
||||
case <-done:
|
||||
return
|
||||
case <-ticker.C:
|
||||
_ = conn.SetWriteDeadline(time.Now().Add(controlWriteTimeout))
|
||||
if err := WriteControlPing(conn); err != nil {
|
||||
_ = conn.Close()
|
||||
if err := session.writeControl(controlFrame{
|
||||
Type: controlFramePing,
|
||||
Version: controlProtocolVersion,
|
||||
}, nil); err != nil {
|
||||
_ = session.conn.Close()
|
||||
return
|
||||
}
|
||||
_ = conn.SetWriteDeadline(time.Time{})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -548,7 +598,21 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, e
|
||||
}()
|
||||
stopCloseOnCancel := closeConnOnContextDone(ctx, conn)
|
||||
defer stopCloseOnCancel()
|
||||
if err := WriteOpReqImport(conn, busid); err != nil {
|
||||
lease, err := c.requestImportLease(ctx, busid)
|
||||
if err != nil {
|
||||
return -1, err
|
||||
}
|
||||
expectedReply := OpRepImport
|
||||
if lease.Valid {
|
||||
expectedReply = OpRepImportExt
|
||||
if err := WriteOpReqImportExt(conn, ImportExtRequest{
|
||||
BusID: busid,
|
||||
LeaseID: lease.ID,
|
||||
ClientNonce: lease.ClientNonce,
|
||||
}); err != nil {
|
||||
return -1, E.Cause(err, "write OP_REQ_IMPORT_EXT")
|
||||
}
|
||||
} else if err := WriteOpReqImport(conn, busid); err != nil {
|
||||
return -1, E.Cause(err, "write OP_REQ_IMPORT")
|
||||
}
|
||||
header, err := ReadOpHeader(conn)
|
||||
@@ -558,7 +622,7 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, e
|
||||
if header.Version != ProtocolVersion {
|
||||
return -1, E.New("unexpected reply version 0x", hex16(header.Version))
|
||||
}
|
||||
if header.Code != OpRepImport {
|
||||
if header.Code != expectedReply {
|
||||
return -1, E.New("unexpected reply code 0x", hex16(header.Code))
|
||||
}
|
||||
if header.Status != OpStatusOK {
|
||||
|
||||
@@ -2,34 +2,130 @@ package usbip
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"reflect"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
E "github.com/sagernet/sing/common/exceptions"
|
||||
)
|
||||
|
||||
const (
|
||||
controlProtocolVersion uint8 = 1
|
||||
|
||||
controlFrameHello uint8 = 1
|
||||
controlFrameAck uint8 = 2
|
||||
controlFrameChanged uint8 = 3
|
||||
controlFramePing uint8 = 4
|
||||
controlFramePong uint8 = 5
|
||||
controlFrameHello uint8 = 1
|
||||
controlFrameAck uint8 = 2
|
||||
controlFrameChanged uint8 = 3
|
||||
controlFramePing uint8 = 4
|
||||
controlFramePong uint8 = 5
|
||||
controlFrameDeviceSnapshot uint8 = 6
|
||||
controlFrameDeviceDelta uint8 = 7
|
||||
controlFrameLeaseRequest uint8 = 8
|
||||
controlFrameLeaseResponse uint8 = 9
|
||||
|
||||
controlCapabilityChanged uint32 = 1 << 0
|
||||
controlCapabilityPingPong uint32 = 1 << 1
|
||||
controlCapabilities = controlCapabilityChanged | controlCapabilityPingPong
|
||||
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
|
||||
controlCapabilities = controlRequiredCapabilities | controlExtensionCapabilities
|
||||
|
||||
controlPrefaceSize = 8
|
||||
controlFrameSize = 16
|
||||
controlPrefaceSize = 8
|
||||
controlFrameSize = 16
|
||||
maxControlPayloadLength = 64<<10 - 1
|
||||
|
||||
deviceStateAvailable = "available"
|
||||
deviceStateBusy = "busy"
|
||||
deviceStateUnavailable = "unavailable"
|
||||
|
||||
importLeaseTTL = 10 * time.Second
|
||||
)
|
||||
|
||||
var controlPreface = [controlPrefaceSize]byte{'S', 'B', 'U', 'S', 'B', 'I', 'P', '1'}
|
||||
|
||||
type controlFrame struct {
|
||||
Type uint8
|
||||
Version uint8
|
||||
_ uint16
|
||||
Capabilities uint32
|
||||
Sequence uint64
|
||||
Type uint8
|
||||
Version uint8
|
||||
PayloadLength uint16
|
||||
Capabilities uint32
|
||||
Sequence uint64
|
||||
}
|
||||
|
||||
type controlMessage struct {
|
||||
Frame controlFrame
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
type controlOutboundMessage struct {
|
||||
Frame controlFrame
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
type DeviceInterfaceV2 struct {
|
||||
Class uint8 `json:"class"`
|
||||
SubClass uint8 `json:"subclass"`
|
||||
Protocol uint8 `json:"protocol"`
|
||||
}
|
||||
|
||||
type DeviceInfoV2 struct {
|
||||
BusID string `json:"busid"`
|
||||
StableID string `json:"stable_id,omitempty"`
|
||||
Backend string `json:"backend,omitempty"`
|
||||
Path string `json:"path,omitempty"`
|
||||
Serial string `json:"serial,omitempty"`
|
||||
VendorID uint16 `json:"vendor_id"`
|
||||
ProductID uint16 `json:"product_id"`
|
||||
BCDDevice uint16 `json:"bcd_device,omitempty"`
|
||||
Speed uint32 `json:"speed"`
|
||||
DeviceClass uint8 `json:"device_class"`
|
||||
DeviceSubClass uint8 `json:"device_subclass"`
|
||||
DeviceProtocol uint8 `json:"device_protocol"`
|
||||
ConfigurationValue uint8 `json:"configuration_value"`
|
||||
NumConfigurations uint8 `json:"num_configurations"`
|
||||
NumInterfaces uint8 `json:"num_interfaces"`
|
||||
Interfaces []DeviceInterfaceV2 `json:"interfaces,omitempty"`
|
||||
State string `json:"state"`
|
||||
StatusCode int `json:"status_code,omitempty"`
|
||||
StatusReason string `json:"status_reason,omitempty"`
|
||||
}
|
||||
|
||||
type controlDeviceSnapshot struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Devices []DeviceInfoV2 `json:"devices"`
|
||||
}
|
||||
|
||||
type controlDeviceDelta struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Added []DeviceInfoV2 `json:"added,omitempty"`
|
||||
Updated []DeviceInfoV2 `json:"updated,omitempty"`
|
||||
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"`
|
||||
}
|
||||
|
||||
type serverImportLease struct {
|
||||
ID uint64
|
||||
SubscriberID uint64
|
||||
BusID string
|
||||
ClientNonce uint64
|
||||
Generation uint64
|
||||
Expires time.Time
|
||||
}
|
||||
|
||||
func WriteControlPreface(w io.Writer) error {
|
||||
@@ -58,10 +154,14 @@ func WriteControlHello(w io.Writer) error {
|
||||
}
|
||||
|
||||
func WriteControlAck(w io.Writer, sequence uint64) error {
|
||||
return writeControlAckWithCapabilities(w, sequence, controlCapabilities)
|
||||
}
|
||||
|
||||
func writeControlAckWithCapabilities(w io.Writer, sequence uint64, capabilities uint32) error {
|
||||
return writeControlFrame(w, controlFrame{
|
||||
Type: controlFrameAck,
|
||||
Version: controlProtocolVersion,
|
||||
Capabilities: controlCapabilities,
|
||||
Capabilities: capabilities,
|
||||
Sequence: sequence,
|
||||
})
|
||||
}
|
||||
@@ -89,24 +189,220 @@ func WriteControlPong(w io.Writer) error {
|
||||
}
|
||||
|
||||
func ReadControlFrame(r io.Reader) (controlFrame, error) {
|
||||
var raw [controlFrameSize]byte
|
||||
if _, err := io.ReadFull(r, raw[:]); err != nil {
|
||||
message, err := readControlMessage(r)
|
||||
if err != nil {
|
||||
return controlFrame{}, err
|
||||
}
|
||||
return controlFrame{
|
||||
Type: raw[0],
|
||||
Version: raw[1],
|
||||
Capabilities: binary.BigEndian.Uint32(raw[4:8]),
|
||||
Sequence: binary.BigEndian.Uint64(raw[8:16]),
|
||||
}, nil
|
||||
if len(message.Payload) > 0 {
|
||||
return controlFrame{}, E.New("unexpected control payload length ", len(message.Payload))
|
||||
}
|
||||
return message.Frame, nil
|
||||
}
|
||||
|
||||
func readControlMessage(r io.Reader) (controlMessage, error) {
|
||||
var raw [controlFrameSize]byte
|
||||
if _, err := io.ReadFull(r, raw[:]); err != nil {
|
||||
return controlMessage{}, err
|
||||
}
|
||||
frame := controlFrame{
|
||||
Type: raw[0],
|
||||
Version: raw[1],
|
||||
PayloadLength: binary.BigEndian.Uint16(raw[2:4]),
|
||||
Capabilities: binary.BigEndian.Uint32(raw[4:8]),
|
||||
Sequence: binary.BigEndian.Uint64(raw[8:16]),
|
||||
}
|
||||
var payload []byte
|
||||
if frame.PayloadLength > 0 {
|
||||
payload = make([]byte, frame.PayloadLength)
|
||||
if _, err := io.ReadFull(r, payload); err != nil {
|
||||
return controlMessage{}, err
|
||||
}
|
||||
}
|
||||
return controlMessage{Frame: frame, Payload: payload}, nil
|
||||
}
|
||||
|
||||
func writeControlFrame(w io.Writer, frame controlFrame) error {
|
||||
return writeControlMessage(w, frame, nil)
|
||||
}
|
||||
|
||||
func writeControlMessage(w io.Writer, frame controlFrame, payload any) error {
|
||||
rawPayload, err := marshalControlPayload(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(rawPayload) > maxControlPayloadLength {
|
||||
return E.New("control payload too large: ", len(rawPayload))
|
||||
}
|
||||
frame.PayloadLength = uint16(len(rawPayload))
|
||||
var raw [controlFrameSize]byte
|
||||
raw[0] = frame.Type
|
||||
raw[1] = frame.Version
|
||||
binary.BigEndian.PutUint16(raw[2:4], frame.PayloadLength)
|
||||
binary.BigEndian.PutUint32(raw[4:8], frame.Capabilities)
|
||||
binary.BigEndian.PutUint64(raw[8:16], frame.Sequence)
|
||||
_, err := w.Write(raw[:])
|
||||
if _, err := w.Write(raw[:]); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(rawPayload) == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err = w.Write(rawPayload)
|
||||
return err
|
||||
}
|
||||
|
||||
func marshalControlPayload(payload any) ([]byte, error) {
|
||||
switch value := payload.(type) {
|
||||
case nil:
|
||||
return nil, nil
|
||||
case []byte:
|
||||
return value, nil
|
||||
default:
|
||||
return json.Marshal(value)
|
||||
}
|
||||
}
|
||||
|
||||
func unmarshalControlPayload(payload []byte, value any) error {
|
||||
if len(payload) == 0 {
|
||||
return E.New("missing control payload")
|
||||
}
|
||||
return json.Unmarshal(payload, value)
|
||||
}
|
||||
|
||||
func negotiatedControlCapabilities(peer uint32) uint32 {
|
||||
return peer & controlCapabilities
|
||||
}
|
||||
|
||||
func supportsControlExtensions(capabilities uint32) bool {
|
||||
return capabilities&controlExtensionCapabilities == controlExtensionCapabilities
|
||||
}
|
||||
|
||||
func deviceInfoV2FromEntry(entry DeviceEntry, backend string, stableID string, state string, statusCode int, statusReason string) DeviceInfoV2 {
|
||||
interfaces := make([]DeviceInterfaceV2, len(entry.Interfaces))
|
||||
for i := range entry.Interfaces {
|
||||
interfaces[i] = DeviceInterfaceV2{
|
||||
Class: entry.Interfaces[i].BInterfaceClass,
|
||||
SubClass: entry.Interfaces[i].BInterfaceSubClass,
|
||||
Protocol: entry.Interfaces[i].BInterfaceProtocol,
|
||||
}
|
||||
}
|
||||
if state == "" {
|
||||
state = deviceStateAvailable
|
||||
}
|
||||
return DeviceInfoV2{
|
||||
BusID: entry.Info.BusIDString(),
|
||||
StableID: stableID,
|
||||
Backend: backend,
|
||||
Path: entry.Info.PathString(),
|
||||
Serial: entry.Info.SerialString(),
|
||||
VendorID: entry.Info.IDVendor,
|
||||
ProductID: entry.Info.IDProduct,
|
||||
BCDDevice: entry.Info.BCDDevice,
|
||||
Speed: entry.Info.Speed,
|
||||
DeviceClass: entry.Info.BDeviceClass,
|
||||
DeviceSubClass: entry.Info.BDeviceSubClass,
|
||||
DeviceProtocol: entry.Info.BDeviceProtocol,
|
||||
ConfigurationValue: entry.Info.BConfigurationValue,
|
||||
NumConfigurations: entry.Info.BNumConfigurations,
|
||||
NumInterfaces: entry.Info.BNumInterfaces,
|
||||
Interfaces: interfaces,
|
||||
State: state,
|
||||
StatusCode: statusCode,
|
||||
StatusReason: statusReason,
|
||||
}
|
||||
}
|
||||
|
||||
func (d DeviceInfoV2) toDeviceEntry() DeviceEntry {
|
||||
var info DeviceInfoTruncated
|
||||
encodePathField(&info.Path, d.Path, d.Serial)
|
||||
copy(info.BusID[:], d.BusID)
|
||||
info.Speed = d.Speed
|
||||
info.IDVendor = d.VendorID
|
||||
info.IDProduct = d.ProductID
|
||||
info.BCDDevice = d.BCDDevice
|
||||
info.BDeviceClass = d.DeviceClass
|
||||
info.BDeviceSubClass = d.DeviceSubClass
|
||||
info.BDeviceProtocol = d.DeviceProtocol
|
||||
info.BConfigurationValue = d.ConfigurationValue
|
||||
info.BNumConfigurations = d.NumConfigurations
|
||||
info.BNumInterfaces = d.NumInterfaces
|
||||
interfaces := make([]DeviceInterface, len(d.Interfaces))
|
||||
for i := range d.Interfaces {
|
||||
interfaces[i] = DeviceInterface{
|
||||
BInterfaceClass: d.Interfaces[i].Class,
|
||||
BInterfaceSubClass: d.Interfaces[i].SubClass,
|
||||
BInterfaceProtocol: d.Interfaces[i].Protocol,
|
||||
}
|
||||
}
|
||||
return DeviceEntry{Info: info, Interfaces: interfaces}
|
||||
}
|
||||
|
||||
func (d DeviceInfoV2) key() DeviceKey {
|
||||
return DeviceKey{
|
||||
BusID: d.BusID,
|
||||
VendorID: d.VendorID,
|
||||
ProductID: d.ProductID,
|
||||
Serial: d.Serial,
|
||||
}
|
||||
}
|
||||
|
||||
func (d DeviceInfoV2) available() bool {
|
||||
return d.State == "" || d.State == deviceStateAvailable
|
||||
}
|
||||
|
||||
func deviceInfoV2Map(devices []DeviceInfoV2) map[string]DeviceInfoV2 {
|
||||
out := make(map[string]DeviceInfoV2, len(devices))
|
||||
for _, device := range devices {
|
||||
if device.BusID == "" {
|
||||
continue
|
||||
}
|
||||
out[device.BusID] = device
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func sortedDeviceInfoV2Values(devices map[string]DeviceInfoV2) []DeviceInfoV2 {
|
||||
busids := make([]string, 0, len(devices))
|
||||
for busid := range devices {
|
||||
busids = append(busids, busid)
|
||||
}
|
||||
sort.Strings(busids)
|
||||
out := make([]DeviceInfoV2, 0, len(busids))
|
||||
for _, busid := range busids {
|
||||
out = append(out, devices[busid])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func deviceInfoV2ToEntries(devices []DeviceInfoV2, availableOnly bool) []DeviceEntry {
|
||||
entries := make([]DeviceEntry, 0, len(devices))
|
||||
for _, device := range devices {
|
||||
if availableOnly && !device.available() {
|
||||
continue
|
||||
}
|
||||
entries = append(entries, device.toDeviceEntry())
|
||||
}
|
||||
return entries
|
||||
}
|
||||
|
||||
func buildControlDeviceDelta(sequence uint64, previous map[string]DeviceInfoV2, current map[string]DeviceInfoV2) controlDeviceDelta {
|
||||
delta := controlDeviceDelta{Sequence: sequence}
|
||||
for busid, device := range current {
|
||||
prev, ok := previous[busid]
|
||||
if !ok {
|
||||
delta.Added = append(delta.Added, device)
|
||||
continue
|
||||
}
|
||||
if !reflect.DeepEqual(prev, device) {
|
||||
delta.Updated = append(delta.Updated, device)
|
||||
}
|
||||
}
|
||||
for busid := range previous {
|
||||
if _, ok := current[busid]; !ok {
|
||||
delta.Removed = append(delta.Removed, busid)
|
||||
}
|
||||
}
|
||||
sort.Slice(delta.Added, func(i, j int) bool { return delta.Added[i].BusID < delta.Added[j].BusID })
|
||||
sort.Slice(delta.Updated, func(i, j int) bool { return delta.Updated[i].BusID < delta.Updated[j].BusID })
|
||||
sort.Strings(delta.Removed)
|
||||
return delta
|
||||
}
|
||||
|
||||
@@ -377,26 +377,55 @@ func (s *darwinFakeUSBIPServer) handleConn(conn net.Conn) {
|
||||
_ = WriteOpRepDevList(conn, []DeviceEntry{s.entry})
|
||||
case OpReqImport:
|
||||
s.handleImport(conn)
|
||||
case OpReqImportExt:
|
||||
s.handleImportExt(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *darwinFakeUSBIPServer) handleControlConn(conn net.Conn) {
|
||||
hello, err := ReadControlFrame(conn)
|
||||
helloMessage, err := readControlMessage(conn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
hello := helloMessage.Frame
|
||||
if hello.Type != controlFrameHello || hello.Version != controlProtocolVersion {
|
||||
return
|
||||
}
|
||||
if err := WriteControlAck(conn, 0); err != nil {
|
||||
capabilities := negotiatedControlCapabilities(hello.Capabilities)
|
||||
if err := writeControlAckWithCapabilities(conn, 0, capabilities); err != nil {
|
||||
return
|
||||
}
|
||||
if supportsControlExtensions(capabilities) {
|
||||
_ = writeControlMessage(conn, controlFrame{
|
||||
Type: controlFrameDeviceSnapshot,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlDeviceSnapshot{
|
||||
Devices: []DeviceInfoV2{deviceInfoV2FromEntry(s.entry, "darwin-fake", "darwin-fake:"+s.entry.Info.BusIDString(), deviceStateAvailable, 0, "available")},
|
||||
})
|
||||
}
|
||||
for {
|
||||
frame, err := ReadControlFrame(conn)
|
||||
message, err := readControlMessage(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
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := WriteControlPong(conn); err != nil {
|
||||
@@ -421,6 +450,22 @@ 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 {
|
||||
_ = WriteOpRepImportExt(conn, OpStatusError, nil)
|
||||
return
|
||||
}
|
||||
info := s.entry.Info
|
||||
if err := WriteOpRepImportExt(conn, OpStatusOK, &info); err != nil {
|
||||
return
|
||||
}
|
||||
s.handleDataSession(conn)
|
||||
}
|
||||
|
||||
func (s *darwinFakeUSBIPServer) handleDataSession(conn net.Conn) {
|
||||
for {
|
||||
header, err := ReadDataHeader(conn)
|
||||
|
||||
+237
-3
@@ -1226,6 +1226,13 @@ func TestServerDispatchConnHandlesControlPingAndChanged(t *testing.T) {
|
||||
require.Equal(t, controlCapabilities, ack.Capabilities)
|
||||
require.Zero(t, ack.Sequence)
|
||||
|
||||
snapshotMessage, err := readControlMessage(conn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, controlFrameDeviceSnapshot, snapshotMessage.Frame.Type)
|
||||
var snapshot controlDeviceSnapshot
|
||||
require.NoError(t, unmarshalControlPayload(snapshotMessage.Payload, &snapshot))
|
||||
require.Empty(t, snapshot.Devices)
|
||||
|
||||
require.NoError(t, WriteControlPing(conn))
|
||||
pong, err := ReadControlFrame(conn)
|
||||
require.NoError(t, err)
|
||||
@@ -1233,10 +1240,108 @@ func TestServerDispatchConnHandlesControlPingAndChanged(t *testing.T) {
|
||||
require.Equal(t, controlProtocolVersion, pong.Version)
|
||||
|
||||
server.broadcastChanged()
|
||||
changed, err := ReadControlFrame(conn)
|
||||
changed, err := readControlMessage(conn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, controlFrameChanged, changed.Type)
|
||||
require.Equal(t, uint64(1), changed.Sequence)
|
||||
require.Equal(t, controlFrameDeviceDelta, changed.Frame.Type)
|
||||
require.Equal(t, uint64(1), changed.Frame.Sequence)
|
||||
var delta controlDeviceDelta
|
||||
require.NoError(t, unmarshalControlPayload(changed.Payload, &delta))
|
||||
require.Equal(t, uint64(1), delta.Sequence)
|
||||
}
|
||||
|
||||
func TestServerControlLeaseEnablesImportExt(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
device := newTestDevice("1-1", 0x1d6b, 0x0002, "serial-1", SpeedHigh)
|
||||
store := newTestDeviceStore(device)
|
||||
store.setStatus("1-1", usbipStatusAvailable)
|
||||
|
||||
serverOps := newTestUSBIPOps(t)
|
||||
serverOps.readUsbipStatus = store.readUsbipStatus
|
||||
serverOps.readSysfsDevice = store.readSysfsDevice
|
||||
serverOps.writeUsbipSockfd = store.writeUsbipSockfd
|
||||
|
||||
server := &ServerService{
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
logger: newTestLogger(),
|
||||
exports: map[string]serverExport{"1-1": {busid: "1-1"}},
|
||||
controlSubs: make(map[uint64]*serverControlConn),
|
||||
controlState: make(map[string]DeviceInfoV2),
|
||||
leases: make(map[uint64]serverImportLease),
|
||||
leaseByBusID: make(map[string]uint64),
|
||||
ops: serverOps,
|
||||
}
|
||||
server.refreshControlState()
|
||||
serverAddr, closeServer := startDispatchServer(t, server)
|
||||
defer closeServer()
|
||||
|
||||
controlConn, err := net.Dial("tcp", serverAddr.String())
|
||||
require.NoError(t, err)
|
||||
defer controlConn.Close()
|
||||
require.NoError(t, WriteControlPreface(controlConn))
|
||||
require.NoError(t, WriteControlHello(controlConn))
|
||||
ack, err := ReadControlFrame(controlConn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, controlCapabilities, ack.Capabilities)
|
||||
_, err = readControlMessage(controlConn)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, writeControlMessage(controlConn, controlFrame{
|
||||
Type: controlFrameLeaseRequest,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlLeaseRequest{BusID: "1-1", ClientNonce: 42}))
|
||||
leaseMessage, err := readControlMessage(controlConn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, controlFrameLeaseResponse, leaseMessage.Frame.Type)
|
||||
var lease controlLeaseResponse
|
||||
require.NoError(t, unmarshalControlPayload(leaseMessage.Payload, &lease))
|
||||
require.Empty(t, lease.ErrorCode)
|
||||
require.Equal(t, uint64(42), lease.ClientNonce)
|
||||
require.NotZero(t, lease.LeaseID)
|
||||
|
||||
require.NoError(t, writeControlMessage(controlConn, controlFrame{
|
||||
Type: controlFrameLeaseRequest,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlLeaseRequest{BusID: "1-1", ClientNonce: 43}))
|
||||
busyMessage, err := readControlMessage(controlConn)
|
||||
require.NoError(t, err)
|
||||
var busy controlLeaseResponse
|
||||
require.NoError(t, unmarshalControlPayload(busyMessage.Payload, &busy))
|
||||
require.Equal(t, "busy", busy.ErrorCode)
|
||||
|
||||
importConn, err := net.Dial("tcp", serverAddr.String())
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, WriteOpReqImportExt(importConn, ImportExtRequest{
|
||||
BusID: "1-1",
|
||||
LeaseID: lease.LeaseID,
|
||||
ClientNonce: lease.ClientNonce,
|
||||
}))
|
||||
header, err := ReadOpHeader(importConn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, OpRepImportExt, header.Code)
|
||||
require.Equal(t, OpStatusOK, header.Status)
|
||||
info, err := ReadOpRepImportBody(importConn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "1-1", info.BusIDString())
|
||||
require.NoError(t, importConn.Close())
|
||||
require.Positive(t, store.lastSockfd("1-1"))
|
||||
|
||||
reuseConn, err := net.Dial("tcp", serverAddr.String())
|
||||
require.NoError(t, err)
|
||||
defer reuseConn.Close()
|
||||
require.NoError(t, WriteOpReqImportExt(reuseConn, ImportExtRequest{
|
||||
BusID: "1-1",
|
||||
LeaseID: lease.LeaseID,
|
||||
ClientNonce: lease.ClientNonce,
|
||||
}))
|
||||
header, err = ReadOpHeader(reuseConn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, OpRepImportExt, header.Code)
|
||||
require.Equal(t, OpStatusError, header.Status)
|
||||
}
|
||||
|
||||
func TestClientAttemptAttachUsesImportReplyAndVHCIAttach(t *testing.T) {
|
||||
@@ -1301,6 +1406,135 @@ func TestClientAttemptAttachUsesImportReplyAndVHCIAttach(t *testing.T) {
|
||||
require.Positive(t, store.lastSockfd("1-1"))
|
||||
}
|
||||
|
||||
func TestClientAttemptAttachUsesImportExtLease(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
controlClient, controlServer := net.Pipe()
|
||||
defer controlClient.Close()
|
||||
defer controlServer.Close()
|
||||
|
||||
controlSession := newClientControlSession(controlClient, controlCapabilities)
|
||||
controlErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
message, err := readControlMessage(controlServer)
|
||||
if err != nil {
|
||||
controlErrCh <- err
|
||||
return
|
||||
}
|
||||
if message.Frame.Type != controlFrameLeaseRequest {
|
||||
controlErrCh <- fmt.Errorf("unexpected control frame %d", message.Frame.Type)
|
||||
return
|
||||
}
|
||||
var request controlLeaseRequest
|
||||
if err := unmarshalControlPayload(message.Payload, &request); err != nil {
|
||||
controlErrCh <- err
|
||||
return
|
||||
}
|
||||
if request.BusID != "1-1" {
|
||||
controlErrCh <- fmt.Errorf("unexpected lease busid %s", request.BusID)
|
||||
return
|
||||
}
|
||||
controlErrCh <- writeControlMessage(controlServer, controlFrame{
|
||||
Type: controlFrameLeaseResponse,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlLeaseResponse{
|
||||
BusID: request.BusID,
|
||||
LeaseID: 55,
|
||||
ClientNonce: request.ClientNonce,
|
||||
Generation: 2,
|
||||
TTLMillis: int64(importLeaseTTL / time.Millisecond),
|
||||
})
|
||||
}()
|
||||
deliverErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
message, err := readControlMessage(controlClient)
|
||||
if err != nil {
|
||||
deliverErrCh <- err
|
||||
return
|
||||
}
|
||||
if message.Frame.Type != controlFrameLeaseResponse {
|
||||
deliverErrCh <- fmt.Errorf("unexpected control response %d", message.Frame.Type)
|
||||
return
|
||||
}
|
||||
var response controlLeaseResponse
|
||||
if err := unmarshalControlPayload(message.Payload, &response); err != nil {
|
||||
deliverErrCh <- err
|
||||
return
|
||||
}
|
||||
controlSession.deliverLeaseResponse(response)
|
||||
deliverErrCh <- nil
|
||||
}()
|
||||
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
defer listener.Close()
|
||||
|
||||
device := newTestDevice("1-1", 0x1d6b, 0x0002, "serial-1", SpeedHigh)
|
||||
serverErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
conn, acceptErr := listener.Accept()
|
||||
if acceptErr != nil {
|
||||
serverErrCh <- acceptErr
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
header, readErr := ReadOpHeader(conn)
|
||||
if readErr != nil {
|
||||
serverErrCh <- readErr
|
||||
return
|
||||
}
|
||||
if header.Code != OpReqImportExt {
|
||||
serverErrCh <- fmt.Errorf("unexpected request code 0x%s", hex16(header.Code))
|
||||
return
|
||||
}
|
||||
request, readErr := ReadOpReqImportExtBody(conn)
|
||||
if readErr != nil {
|
||||
serverErrCh <- readErr
|
||||
return
|
||||
}
|
||||
if request.BusID != "1-1" || request.LeaseID != 55 || request.ClientNonce != 1 {
|
||||
serverErrCh <- fmt.Errorf("unexpected import-ext request %+v", request)
|
||||
return
|
||||
}
|
||||
info := device.toProtocol()
|
||||
serverErrCh <- WriteOpRepImportExt(conn, OpStatusOK, &info)
|
||||
}()
|
||||
|
||||
ops := newTestUSBIPOps(t)
|
||||
ops.vhciPickFreePort = func(speed uint32) (int, error) {
|
||||
require.Equal(t, SpeedHigh, speed)
|
||||
return 4, nil
|
||||
}
|
||||
ops.vhciAttach = func(port int, _ uintptr, devid uint32, speed uint32) error {
|
||||
require.Equal(t, 4, port)
|
||||
info := device.toProtocol()
|
||||
require.Equal(t, info.DevID(), devid)
|
||||
require.Equal(t, SpeedHigh, speed)
|
||||
return nil
|
||||
}
|
||||
|
||||
client := &ClientService{
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
logger: newTestLogger(),
|
||||
dialer: testDialer{},
|
||||
serverAddr: M.SocksaddrFromNet(listener.Addr()),
|
||||
ops: ops,
|
||||
}
|
||||
client.setControlSession(controlSession)
|
||||
defer client.clearControlSession(controlSession, errClientControlSessionClosed)
|
||||
|
||||
port, err := client.attemptAttach(ctx, "1-1")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 4, port)
|
||||
require.NoError(t, <-controlErrCh)
|
||||
require.NoError(t, <-deliverErrCh)
|
||||
require.NoError(t, <-serverErrCh)
|
||||
}
|
||||
|
||||
func TestClientAttemptAttachWithOpaqueConnRelay(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -14,10 +14,12 @@ const (
|
||||
|
||||
ProtocolVersion uint16 = 0x0111
|
||||
|
||||
OpReqDevList uint16 = 0x8005
|
||||
OpRepDevList uint16 = 0x0005
|
||||
OpReqImport uint16 = 0x8003
|
||||
OpRepImport uint16 = 0x0003
|
||||
OpReqDevList uint16 = 0x8005
|
||||
OpRepDevList uint16 = 0x0005
|
||||
OpReqImport uint16 = 0x8003
|
||||
OpRepImport uint16 = 0x0003
|
||||
OpReqImportExt uint16 = 0x8f03
|
||||
OpRepImportExt uint16 = 0x0f03
|
||||
|
||||
OpStatusOK uint32 = 0
|
||||
OpStatusError uint32 = 1
|
||||
@@ -26,6 +28,7 @@ const (
|
||||
maxOpRepDevListBodyBytes = 8 << 20
|
||||
deviceInfoWireSize = 312
|
||||
deviceInterfaceWireSize = 4
|
||||
importExtBodyWireSize = 56
|
||||
)
|
||||
|
||||
// USB speeds (enum usb_device_speed).
|
||||
@@ -79,6 +82,13 @@ type DeviceEntry struct {
|
||||
Interfaces []DeviceInterface
|
||||
}
|
||||
|
||||
type ImportExtRequest struct {
|
||||
BusID string
|
||||
LeaseID uint64
|
||||
ClientNonce uint64
|
||||
Flags uint32
|
||||
}
|
||||
|
||||
// WriteOpHeader emits the 8-byte OP header.
|
||||
func WriteOpHeader(w io.Writer, code uint16, status uint32) error {
|
||||
return binary.Write(w, binary.BigEndian, OpHeader{
|
||||
@@ -118,6 +128,22 @@ func WriteOpReqImport(w io.Writer, busid string) error {
|
||||
return binary.Write(w, binary.BigEndian, field)
|
||||
}
|
||||
|
||||
func WriteOpReqImportExt(w io.Writer, request ImportExtRequest) error {
|
||||
if err := WriteOpHeader(w, OpReqImportExt, OpStatusOK); 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
|
||||
}
|
||||
|
||||
// ReadOpReqImportBody reads the 32-byte busid that follows the OP header.
|
||||
func ReadOpReqImportBody(r io.Reader) (string, error) {
|
||||
var field [32]byte
|
||||
@@ -127,9 +153,30 @@ func ReadOpReqImportBody(r io.Reader) (string, error) {
|
||||
return cstring(field[:]), nil
|
||||
}
|
||||
|
||||
func ReadOpReqImportExtBody(r io.Reader) (ImportExtRequest, error) {
|
||||
var raw [importExtBodyWireSize]byte
|
||||
if _, err := io.ReadFull(r, raw[:]); 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
|
||||
}
|
||||
|
||||
// WriteOpRepImport sends OP_REP_IMPORT. If status != OpStatusOK, info is omitted.
|
||||
func WriteOpRepImport(w io.Writer, status uint32, info *DeviceInfoTruncated) error {
|
||||
if err := WriteOpHeader(w, OpRepImport, status); err != nil {
|
||||
return writeOpRepImport(w, OpRepImport, status, info)
|
||||
}
|
||||
|
||||
func WriteOpRepImportExt(w io.Writer, status uint32, info *DeviceInfoTruncated) error {
|
||||
return writeOpRepImport(w, OpRepImportExt, status, info)
|
||||
}
|
||||
|
||||
func writeOpRepImport(w io.Writer, code uint16, status uint32, info *DeviceInfoTruncated) error {
|
||||
if err := WriteOpHeader(w, code, status); err != nil {
|
||||
return err
|
||||
}
|
||||
if status != OpStatusOK {
|
||||
|
||||
@@ -89,6 +89,57 @@ func TestControlPrefaceAndFrames(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestControlMessagePayloadRoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
payload := controlDeviceSnapshot{
|
||||
Sequence: 3,
|
||||
Devices: []DeviceInfoV2{{
|
||||
BusID: "1-2",
|
||||
StableID: "usb:1d6b:0002:serial-1",
|
||||
Backend: "linux-sysfs",
|
||||
VendorID: 0x1d6b,
|
||||
ProductID: 0x0002,
|
||||
Speed: SpeedHigh,
|
||||
State: deviceStateAvailable,
|
||||
}},
|
||||
}
|
||||
var buffer bytes.Buffer
|
||||
require.NoError(t, writeControlMessage(&buffer, controlFrame{
|
||||
Type: controlFrameDeviceSnapshot,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: 3,
|
||||
}, payload))
|
||||
|
||||
message, err := readControlMessage(&buffer)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, controlFrameDeviceSnapshot, message.Frame.Type)
|
||||
require.Equal(t, uint64(3), message.Frame.Sequence)
|
||||
require.Positive(t, message.Frame.PayloadLength)
|
||||
|
||||
var decoded controlDeviceSnapshot
|
||||
require.NoError(t, unmarshalControlPayload(message.Payload, &decoded))
|
||||
require.Equal(t, payload, decoded)
|
||||
}
|
||||
|
||||
func TestControlMessagePayloadSizeGuard(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var buffer bytes.Buffer
|
||||
err := writeControlMessage(&buffer, controlFrame{Type: controlFrameDeviceSnapshot}, bytes.Repeat([]byte{'x'}, maxControlPayloadLength+1))
|
||||
require.ErrorContains(t, err, "control payload too large")
|
||||
}
|
||||
|
||||
func TestReadControlFrameRejectsPayload(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var buffer bytes.Buffer
|
||||
require.NoError(t, writeControlMessage(&buffer, controlFrame{Type: controlFrameDeviceSnapshot}, []byte(`{}`)))
|
||||
|
||||
_, err := ReadControlFrame(&buffer)
|
||||
require.ErrorContains(t, err, "unexpected control payload length")
|
||||
}
|
||||
|
||||
func TestOpHeaderRoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -114,6 +165,28 @@ func TestOpHeaderRoundTrip(t *testing.T) {
|
||||
}, ParseOpHeader(raw[:]))
|
||||
}
|
||||
|
||||
func TestOpReqImportExtRoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
request := ImportExtRequest{
|
||||
BusID: "1-2",
|
||||
LeaseID: 9,
|
||||
ClientNonce: 7,
|
||||
Flags: 1,
|
||||
}
|
||||
var buffer bytes.Buffer
|
||||
require.NoError(t, WriteOpReqImportExt(&buffer, request))
|
||||
|
||||
header, err := ReadOpHeader(&buffer)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, OpReqImportExt, header.Code)
|
||||
require.Equal(t, OpStatusOK, header.Status)
|
||||
|
||||
parsed, err := ReadOpReqImportExtBody(&buffer)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, request, parsed)
|
||||
}
|
||||
|
||||
func TestOpReqImportRoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -231,6 +304,44 @@ func TestDeviceInfoHelpers(t *testing.T) {
|
||||
require.Equal(t, uint32(0x00030009), info.DevID())
|
||||
}
|
||||
|
||||
func TestDeviceInfoV2RoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var path [256]byte
|
||||
encodePathField(&path, "/sys/bus/usb/devices/1-2", "serial-1")
|
||||
entry := DeviceEntry{
|
||||
Info: DeviceInfoTruncated{
|
||||
Path: path,
|
||||
BusNum: 1,
|
||||
DevNum: 2,
|
||||
Speed: SpeedSuper,
|
||||
IDVendor: 0x1d6b,
|
||||
IDProduct: 0x0002,
|
||||
BCDDevice: 0x0100,
|
||||
BDeviceClass: 0xff,
|
||||
BDeviceSubClass: 1,
|
||||
BDeviceProtocol: 2,
|
||||
BConfigurationValue: 1,
|
||||
BNumConfigurations: 1,
|
||||
BNumInterfaces: 1,
|
||||
},
|
||||
Interfaces: []DeviceInterface{{BInterfaceClass: 0xff, BInterfaceSubClass: 1, BInterfaceProtocol: 2}},
|
||||
}
|
||||
copy(entry.Info.BusID[:], "1-2")
|
||||
|
||||
info := deviceInfoV2FromEntry(entry, "linux-sysfs", "usb:1d6b:0002:serial-1", deviceStateAvailable, 1, "available")
|
||||
require.Equal(t, "1-2", info.BusID)
|
||||
require.Equal(t, "serial-1", info.Serial)
|
||||
require.True(t, info.available())
|
||||
require.Equal(t, DeviceKey{BusID: "1-2", VendorID: 0x1d6b, ProductID: 0x0002, Serial: "serial-1"}, info.key())
|
||||
roundTrip := info.toDeviceEntry()
|
||||
require.Equal(t, "1-2", roundTrip.Info.BusIDString())
|
||||
require.Equal(t, "serial-1", roundTrip.Info.SerialString())
|
||||
require.Equal(t, uint16(0x1d6b), roundTrip.Info.IDVendor)
|
||||
require.Equal(t, uint16(0x0002), roundTrip.Info.IDProduct)
|
||||
require.Equal(t, entry.Interfaces, roundTrip.Interfaces)
|
||||
}
|
||||
|
||||
func TestEncodePathFieldSkipsSerialWithoutRoom(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
+260
-18
@@ -33,9 +33,10 @@ type serverExport struct {
|
||||
}
|
||||
|
||||
type serverControlConn struct {
|
||||
id uint64
|
||||
conn net.Conn
|
||||
send chan controlFrame
|
||||
id uint64
|
||||
capabilities uint32
|
||||
conn net.Conn
|
||||
send chan controlOutboundMessage
|
||||
}
|
||||
|
||||
type ServerService struct {
|
||||
@@ -54,6 +55,10 @@ type ServerService struct {
|
||||
controlSeq uint64
|
||||
controlNextID uint64
|
||||
controlSubs map[uint64]*serverControlConn
|
||||
controlState map[string]DeviceInfoV2
|
||||
leaseNextID uint64
|
||||
leases map[uint64]serverImportLease
|
||||
leaseByBusID map[string]uint64
|
||||
|
||||
reconcileMu sync.Mutex
|
||||
}
|
||||
@@ -81,7 +86,10 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string,
|
||||
Network: []string{N.NetworkTCP},
|
||||
Listen: options.ListenOptions,
|
||||
}),
|
||||
controlSubs: make(map[uint64]*serverControlConn),
|
||||
controlSubs: make(map[uint64]*serverControlConn),
|
||||
controlState: make(map[string]DeviceInfoV2),
|
||||
leases: make(map[uint64]serverImportLease),
|
||||
leaseByBusID: make(map[string]uint64),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -187,6 +195,8 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error {
|
||||
}
|
||||
if notify && changed {
|
||||
s.broadcastChanged()
|
||||
} else {
|
||||
s.refreshControlState()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -307,6 +317,8 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {
|
||||
s.handleDevList(conn)
|
||||
case OpReqImport:
|
||||
s.handleImport(conn)
|
||||
case OpReqImportExt:
|
||||
s.handleImportExt(conn)
|
||||
default:
|
||||
s.logger.Debug("unknown opcode 0x", hex16(header.Code))
|
||||
}
|
||||
@@ -314,21 +326,26 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {
|
||||
|
||||
func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
hello, err := ReadControlFrame(conn)
|
||||
helloMessage, err := readControlMessage(conn)
|
||||
if err != nil {
|
||||
s.logger.Debug("read control hello: ", err)
|
||||
return
|
||||
}
|
||||
if hello.Type != controlFrameHello || hello.Version != controlProtocolVersion || hello.Capabilities&controlCapabilities != controlCapabilities {
|
||||
hello := helloMessage.Frame
|
||||
if hello.Type != controlFrameHello || hello.Version != controlProtocolVersion || hello.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities {
|
||||
s.logger.Debug("invalid control hello")
|
||||
return
|
||||
}
|
||||
sub, seq := s.registerControlConn(conn)
|
||||
capabilities := negotiatedControlCapabilities(hello.Capabilities)
|
||||
sub, seq := s.registerControlConn(conn, capabilities)
|
||||
defer s.unregisterControlConn(sub.id)
|
||||
if err := WriteControlAck(conn, seq); err != nil {
|
||||
if err := writeControlAckWithCapabilities(conn, seq, capabilities); err != nil {
|
||||
s.logger.Debug("write control ack: ", err)
|
||||
return
|
||||
}
|
||||
if supportsControlExtensions(capabilities) {
|
||||
s.enqueueControlSnapshot(sub, seq)
|
||||
}
|
||||
readDone := make(chan struct{})
|
||||
go s.readControlConn(sub, readDone)
|
||||
for {
|
||||
@@ -337,8 +354,8 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
return
|
||||
case <-readDone:
|
||||
return
|
||||
case frame := <-sub.send:
|
||||
if err := writeControlFrame(conn, frame); err != nil {
|
||||
case message := <-sub.send:
|
||||
if err := writeControlMessage(conn, message.Frame, message.Payload); err != nil {
|
||||
s.logger.Debug("write control frame: ", err)
|
||||
return
|
||||
}
|
||||
@@ -349,13 +366,20 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
func (s *ServerService) readControlConn(sub *serverControlConn, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
for {
|
||||
frame, err := ReadControlFrame(sub.conn)
|
||||
message, err := readControlMessage(sub.conn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
frame := message.Frame
|
||||
switch frame.Type {
|
||||
case controlFramePing:
|
||||
s.enqueueControlFrame(sub, controlFrame{Type: controlFramePong, Version: controlProtocolVersion})
|
||||
case controlFrameLeaseRequest:
|
||||
if supportsControlExtensions(sub.capabilities) {
|
||||
s.handleControlLeaseRequest(sub, message.Payload)
|
||||
continue
|
||||
}
|
||||
return
|
||||
default:
|
||||
return
|
||||
}
|
||||
@@ -379,10 +403,32 @@ func (s *ServerService) handleImport(conn net.Conn) {
|
||||
s.logger.Debug("read import body: ", err)
|
||||
return
|
||||
}
|
||||
s.handleImportBusID(conn, busid, false)
|
||||
}
|
||||
|
||||
func (s *ServerService) handleImportExt(conn net.Conn) {
|
||||
request, err := ReadOpReqImportExtBody(conn)
|
||||
if err != nil {
|
||||
s.logger.Debug("read import-ext body: ", err)
|
||||
return
|
||||
}
|
||||
if !s.consumeImportLease(request) {
|
||||
s.logger.Info("import-ext rejected (invalid lease): ", request.BusID)
|
||||
_ = WriteOpRepImportExt(conn, OpStatusError, nil)
|
||||
return
|
||||
}
|
||||
s.handleImportBusID(conn, request.BusID, true)
|
||||
}
|
||||
|
||||
func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended bool) {
|
||||
writeReply := WriteOpRepImport
|
||||
if extended {
|
||||
writeReply = WriteOpRepImportExt
|
||||
}
|
||||
export, ok := s.claimExport(busid)
|
||||
if !ok {
|
||||
s.logger.Info("import rejected (unknown or busy busid): ", busid)
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return
|
||||
}
|
||||
releaseClaim := true
|
||||
@@ -392,7 +438,7 @@ func (s *ServerService) handleImport(conn net.Conn) {
|
||||
}
|
||||
}()
|
||||
info := export.entry.Info
|
||||
if err := WriteOpRepImport(conn, OpStatusOK, &info); err != nil {
|
||||
if err := writeReply(conn, OpStatusOK, &info); err != nil {
|
||||
s.logger.Warn("reply import ", busid, ": ", err)
|
||||
return
|
||||
}
|
||||
@@ -421,14 +467,15 @@ func (s *ServerService) reconcileLoop() {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) registerControlConn(conn net.Conn) (*serverControlConn, uint64) {
|
||||
func (s *ServerService) registerControlConn(conn net.Conn, capabilities uint32) (*serverControlConn, uint64) {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.controlNextID++
|
||||
sub := &serverControlConn{
|
||||
id: s.controlNextID,
|
||||
conn: conn,
|
||||
send: make(chan controlFrame, 16),
|
||||
id: s.controlNextID,
|
||||
capabilities: capabilities,
|
||||
conn: conn,
|
||||
send: make(chan controlOutboundMessage, 16),
|
||||
}
|
||||
s.controlSubs[sub.id] = sub
|
||||
return sub, s.controlSeq
|
||||
@@ -438,6 +485,7 @@ func (s *ServerService) unregisterControlConn(id uint64) {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
delete(s.controlSubs, id)
|
||||
s.deleteImportLeasesForSubscriberLocked(id)
|
||||
}
|
||||
|
||||
func (s *ServerService) closeControlSubscribers() {
|
||||
@@ -454,9 +502,14 @@ func (s *ServerService) closeControlSubscribers() {
|
||||
}
|
||||
|
||||
func (s *ServerService) broadcastChanged() {
|
||||
devices := s.buildDeviceStateV2()
|
||||
nextState := deviceInfoV2Map(devices)
|
||||
|
||||
s.controlMu.Lock()
|
||||
s.controlSeq++
|
||||
sequence := s.controlSeq
|
||||
delta := buildControlDeviceDelta(sequence, s.controlState, nextState)
|
||||
s.controlState = nextState
|
||||
subs := make([]*serverControlConn, 0, len(s.controlSubs))
|
||||
for _, sub := range s.controlSubs {
|
||||
subs = append(subs, sub)
|
||||
@@ -464,19 +517,208 @@ func (s *ServerService) broadcastChanged() {
|
||||
s.controlMu.Unlock()
|
||||
frame := controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: sequence}
|
||||
for _, sub := range subs {
|
||||
if supportsControlExtensions(sub.capabilities) {
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameDeviceDelta,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
}, delta, frame)
|
||||
continue
|
||||
}
|
||||
s.enqueueControlFrame(sub, frame)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlFrame(sub *serverControlConn, frame controlFrame) {
|
||||
s.enqueueControlMessage(sub, controlOutboundMessage{Frame: frame})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlPayload(sub *serverControlConn, frame controlFrame, payload any, fallback controlFrame) {
|
||||
rawPayload, err := marshalControlPayload(payload)
|
||||
if err != nil || len(rawPayload) > maxControlPayloadLength {
|
||||
s.enqueueControlFrame(sub, fallback)
|
||||
return
|
||||
}
|
||||
s.enqueueControlMessage(sub, controlOutboundMessage{Frame: frame, Payload: rawPayload})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlSnapshot(sub *serverControlConn, sequence uint64) {
|
||||
devices := s.buildDeviceStateV2()
|
||||
s.controlMu.Lock()
|
||||
s.controlState = deviceInfoV2Map(devices)
|
||||
s.controlMu.Unlock()
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameDeviceSnapshot,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
}, controlDeviceSnapshot{Sequence: sequence, Devices: devices}, controlFrame{
|
||||
Type: controlFrameChanged,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlMessage(sub *serverControlConn, message controlOutboundMessage) {
|
||||
select {
|
||||
case sub.send <- frame:
|
||||
case sub.send <- message:
|
||||
default:
|
||||
s.logger.Debug("control subscriber ", sub.id, " lagged behind")
|
||||
_ = sub.conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) refreshControlState() {
|
||||
devices := s.buildDeviceStateV2()
|
||||
s.controlMu.Lock()
|
||||
s.controlState = deviceInfoV2Map(devices)
|
||||
s.controlMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 {
|
||||
exports := s.currentExports()
|
||||
if len(exports) == 0 {
|
||||
return nil
|
||||
}
|
||||
devices := make([]DeviceInfoV2, 0, len(exports))
|
||||
for _, export := range exports {
|
||||
devices = append(devices, deviceInfoV2FromEntry(export.entry, "darwin-iokit", darwinStableID(export.registryID), deviceStateAvailable, 0, "available"))
|
||||
}
|
||||
return devices
|
||||
}
|
||||
|
||||
func (s *ServerService) handleControlLeaseRequest(sub *serverControlConn, payload []byte) {
|
||||
var request controlLeaseRequest
|
||||
if err := unmarshalControlPayload(payload, &request); err != nil {
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameLeaseResponse,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlLeaseResponse{
|
||||
ErrorCode: "bad_request",
|
||||
ErrorMessage: err.Error(),
|
||||
}, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()})
|
||||
return
|
||||
}
|
||||
response := s.createControlLeaseResponse(sub.id, request, s.darwinLeaseAvailable)
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameLeaseResponse,
|
||||
Version: controlProtocolVersion,
|
||||
}, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()})
|
||||
}
|
||||
|
||||
func (s *ServerService) darwinLeaseAvailable(busid string) (bool, string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
export, ok := s.exports[busid]
|
||||
if !ok {
|
||||
return false, "unknown busid"
|
||||
}
|
||||
if export.busy {
|
||||
return false, "busy"
|
||||
}
|
||||
return true, ""
|
||||
}
|
||||
|
||||
func (s *ServerService) createControlLeaseResponse(subID uint64, request controlLeaseRequest, available func(string) (bool, string)) controlLeaseResponse {
|
||||
response := controlLeaseResponse{
|
||||
BusID: request.BusID,
|
||||
ClientNonce: request.ClientNonce,
|
||||
}
|
||||
if request.BusID == "" {
|
||||
response.ErrorCode = "bad_request"
|
||||
response.ErrorMessage = "missing busid"
|
||||
return response
|
||||
}
|
||||
if ok, reason := available(request.BusID); !ok {
|
||||
response.ErrorCode = "unavailable"
|
||||
response.ErrorMessage = reason
|
||||
return response
|
||||
}
|
||||
now := time.Now()
|
||||
expires := now.Add(importLeaseTTL)
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.cleanupExpiredImportLeasesLocked(now)
|
||||
if _, exists := s.leaseByBusID[request.BusID]; exists {
|
||||
response.ErrorCode = "busy"
|
||||
response.ErrorMessage = "lease already active"
|
||||
return response
|
||||
}
|
||||
s.leaseNextID++
|
||||
lease := serverImportLease{
|
||||
ID: s.leaseNextID,
|
||||
SubscriberID: subID,
|
||||
BusID: request.BusID,
|
||||
ClientNonce: request.ClientNonce,
|
||||
Generation: s.controlSeq,
|
||||
Expires: expires,
|
||||
}
|
||||
if s.leases == nil {
|
||||
s.leases = make(map[uint64]serverImportLease)
|
||||
}
|
||||
if s.leaseByBusID == nil {
|
||||
s.leaseByBusID = make(map[string]uint64)
|
||||
}
|
||||
s.leases[lease.ID] = lease
|
||||
s.leaseByBusID[lease.BusID] = lease.ID
|
||||
response.LeaseID = lease.ID
|
||||
response.Generation = lease.Generation
|
||||
response.TTLMillis = int64(importLeaseTTL / time.Millisecond)
|
||||
return response
|
||||
}
|
||||
|
||||
func (s *ServerService) consumeImportLease(request ImportExtRequest) bool {
|
||||
now := time.Now()
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.cleanupExpiredImportLeasesLocked(now)
|
||||
lease, ok := s.leases[request.LeaseID]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if lease.BusID != request.BusID || lease.ClientNonce != request.ClientNonce {
|
||||
return false
|
||||
}
|
||||
delete(s.leases, request.LeaseID)
|
||||
delete(s.leaseByBusID, request.BusID)
|
||||
return now.Before(lease.Expires)
|
||||
}
|
||||
|
||||
func (s *ServerService) cleanupExpiredImportLeasesLocked(now time.Time) {
|
||||
if s.leases == nil {
|
||||
s.leases = make(map[uint64]serverImportLease)
|
||||
}
|
||||
if s.leaseByBusID == nil {
|
||||
s.leaseByBusID = make(map[string]uint64)
|
||||
}
|
||||
for id, lease := range s.leases {
|
||||
if now.Before(lease.Expires) {
|
||||
continue
|
||||
}
|
||||
delete(s.leases, id)
|
||||
delete(s.leaseByBusID, lease.BusID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) deleteImportLeasesForSubscriberLocked(subID uint64) {
|
||||
for id, lease := range s.leases {
|
||||
if lease.SubscriberID != subID {
|
||||
continue
|
||||
}
|
||||
delete(s.leases, id)
|
||||
delete(s.leaseByBusID, lease.BusID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) currentControlSequence() uint64 {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
return s.controlSeq
|
||||
}
|
||||
|
||||
func darwinStableID(registryID uint64) string {
|
||||
return "darwin-registry:" + hex32(uint32(registryID>>32)) + hex32(uint32(registryID))
|
||||
}
|
||||
|
||||
type darwinServerDataSession struct {
|
||||
ctx context.Context
|
||||
logger log.ContextLogger
|
||||
|
||||
+316
-23
@@ -34,9 +34,10 @@ type serverExport struct {
|
||||
}
|
||||
|
||||
type serverControlConn struct {
|
||||
id uint64
|
||||
conn net.Conn
|
||||
send chan controlFrame
|
||||
id uint64
|
||||
capabilities uint32
|
||||
conn net.Conn
|
||||
send chan controlOutboundMessage
|
||||
}
|
||||
|
||||
type ServerService struct {
|
||||
@@ -56,6 +57,10 @@ type ServerService struct {
|
||||
controlSeq uint64
|
||||
controlNextID uint64
|
||||
controlSubs map[uint64]*serverControlConn
|
||||
controlState map[string]DeviceInfoV2
|
||||
leaseNextID uint64
|
||||
leases map[uint64]serverImportLease
|
||||
leaseByBusID map[string]uint64
|
||||
|
||||
reconcileMu sync.Mutex
|
||||
}
|
||||
@@ -83,8 +88,11 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string,
|
||||
Network: []string{N.NetworkTCP},
|
||||
Listen: options.ListenOptions,
|
||||
}),
|
||||
controlSubs: make(map[uint64]*serverControlConn),
|
||||
ops: systemUSBIPOps,
|
||||
controlSubs: make(map[uint64]*serverControlConn),
|
||||
controlState: make(map[string]DeviceInfoV2),
|
||||
leases: make(map[uint64]serverImportLease),
|
||||
leaseByBusID: make(map[string]uint64),
|
||||
ops: systemUSBIPOps,
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
@@ -267,6 +275,8 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error {
|
||||
}
|
||||
if notify && changed {
|
||||
s.broadcastChanged()
|
||||
} else {
|
||||
s.refreshControlState()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -357,6 +367,8 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {
|
||||
s.handleDevList(conn)
|
||||
case OpReqImport:
|
||||
closeConn = !s.handleImport(conn)
|
||||
case OpReqImportExt:
|
||||
closeConn = !s.handleImportExt(conn)
|
||||
default:
|
||||
s.logger.Debug("unknown opcode 0x", hex16(header.Code))
|
||||
}
|
||||
@@ -365,11 +377,12 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {
|
||||
func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
|
||||
hello, err := ReadControlFrame(conn)
|
||||
helloMessage, err := readControlMessage(conn)
|
||||
if err != nil {
|
||||
s.logger.Debug("read control hello: ", err)
|
||||
return
|
||||
}
|
||||
hello := helloMessage.Frame
|
||||
if hello.Type != controlFrameHello {
|
||||
s.logger.Debug("unexpected control frame ", hello.Type, " before hello")
|
||||
return
|
||||
@@ -378,18 +391,22 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
s.logger.Debug("unsupported control version ", hello.Version)
|
||||
return
|
||||
}
|
||||
if hello.Capabilities&controlCapabilities != controlCapabilities {
|
||||
if hello.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities {
|
||||
s.logger.Debug("missing control capabilities 0x", hello.Capabilities)
|
||||
return
|
||||
}
|
||||
capabilities := negotiatedControlCapabilities(hello.Capabilities)
|
||||
|
||||
sub, seq := s.registerControlConn(conn)
|
||||
sub, seq := s.registerControlConn(conn, capabilities)
|
||||
defer s.unregisterControlConn(sub.id)
|
||||
|
||||
if err := WriteControlAck(conn, seq); err != nil {
|
||||
if err := writeControlAckWithCapabilities(conn, seq, capabilities); err != nil {
|
||||
s.logger.Debug("write control ack: ", err)
|
||||
return
|
||||
}
|
||||
if supportsControlExtensions(capabilities) {
|
||||
s.enqueueControlSnapshot(sub, seq)
|
||||
}
|
||||
|
||||
readDone := make(chan struct{})
|
||||
go s.readControlConn(sub, readDone)
|
||||
@@ -399,8 +416,8 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
return
|
||||
case <-readDone:
|
||||
return
|
||||
case frame := <-sub.send:
|
||||
if err := writeControlFrame(conn, frame); err != nil {
|
||||
case message := <-sub.send:
|
||||
if err := writeControlMessage(conn, message.Frame, message.Payload); err != nil {
|
||||
s.logger.Debug("write control frame: ", err)
|
||||
return
|
||||
}
|
||||
@@ -411,16 +428,23 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
|
||||
func (s *ServerService) readControlConn(sub *serverControlConn, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
for {
|
||||
frame, err := ReadControlFrame(sub.conn)
|
||||
message, err := readControlMessage(sub.conn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
frame := message.Frame
|
||||
switch frame.Type {
|
||||
case controlFramePing:
|
||||
s.enqueueControlFrame(sub, controlFrame{
|
||||
Type: controlFramePong,
|
||||
Version: controlProtocolVersion,
|
||||
})
|
||||
case controlFrameLeaseRequest:
|
||||
if supportsControlExtensions(sub.capabilities) {
|
||||
s.handleControlLeaseRequest(sub, message.Payload)
|
||||
continue
|
||||
}
|
||||
return
|
||||
default:
|
||||
return
|
||||
}
|
||||
@@ -468,41 +492,63 @@ func (s *ServerService) handleImport(conn net.Conn) bool {
|
||||
s.logger.Debug("read import body: ", err)
|
||||
return false
|
||||
}
|
||||
return s.handleImportBusID(conn, busid, false)
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
if !s.consumeImportLease(request) {
|
||||
s.logger.Info("import-ext rejected (invalid lease): ", request.BusID)
|
||||
_ = WriteOpRepImportExt(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
return s.handleImportBusID(conn, request.BusID, true)
|
||||
}
|
||||
|
||||
func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended bool) bool {
|
||||
writeReply := WriteOpRepImport
|
||||
if extended {
|
||||
writeReply = WriteOpRepImportExt
|
||||
}
|
||||
if !s.isExported(busid) {
|
||||
s.logger.Info("import rejected (unknown busid): ", busid)
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
status, err := s.ops.readUsbipStatus(busid)
|
||||
if err != nil || status != usbipStatusAvailable {
|
||||
s.logger.Info("import rejected (busid ", busid, " status=", status, " err=", err, ")")
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
dev, err := s.ops.readSysfsDevice(busid, sysBusDevicePath(busid))
|
||||
if err != nil {
|
||||
s.logger.Warn("refresh ", busid, ": ", err)
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
handoff, err := newUSBIPConnHandoff(conn)
|
||||
if err != nil {
|
||||
s.logger.Warn("prepare handoff ", busid, ": ", err)
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
defer handoff.Close()
|
||||
s.logger.Debug("usbip server handoff ", busid, ": ", handoff.mode())
|
||||
if err := s.ops.writeUsbipSockfd(busid, int(handoff.kernelFD())); err != nil {
|
||||
s.logger.Warn("hand off ", busid, " to kernel: ", err)
|
||||
_ = WriteOpRepImport(conn, OpStatusError, nil)
|
||||
_ = writeReply(conn, OpStatusError, nil)
|
||||
return false
|
||||
}
|
||||
if err := handoff.closeKernelFD(); err != nil {
|
||||
s.logger.Debug("close kernel fd ", busid, ": ", err)
|
||||
}
|
||||
info := dev.toProtocol()
|
||||
if err := WriteOpRepImport(conn, OpStatusOK, &info); err != nil {
|
||||
if err := writeReply(conn, OpStatusOK, &info); err != nil {
|
||||
s.logger.Warn("reply import ", busid, ": ", err)
|
||||
_ = s.ops.writeUsbipSockfd(busid, -1)
|
||||
return false
|
||||
@@ -577,14 +623,15 @@ func (s *ServerService) reconcileLoop() {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) registerControlConn(conn net.Conn) (*serverControlConn, uint64) {
|
||||
func (s *ServerService) registerControlConn(conn net.Conn, capabilities uint32) (*serverControlConn, uint64) {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.controlNextID++
|
||||
sub := &serverControlConn{
|
||||
id: s.controlNextID,
|
||||
conn: conn,
|
||||
send: make(chan controlFrame, 16),
|
||||
id: s.controlNextID,
|
||||
capabilities: capabilities,
|
||||
conn: conn,
|
||||
send: make(chan controlOutboundMessage, 16),
|
||||
}
|
||||
s.controlSubs[sub.id] = sub
|
||||
return sub, s.controlSeq
|
||||
@@ -594,6 +641,7 @@ func (s *ServerService) unregisterControlConn(id uint64) {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
delete(s.controlSubs, id)
|
||||
s.deleteImportLeasesForSubscriberLocked(id)
|
||||
}
|
||||
|
||||
func (s *ServerService) closeControlSubscribers() {
|
||||
@@ -610,9 +658,14 @@ func (s *ServerService) closeControlSubscribers() {
|
||||
}
|
||||
|
||||
func (s *ServerService) broadcastChanged() {
|
||||
devices := s.buildDeviceStateV2()
|
||||
nextState := deviceInfoV2Map(devices)
|
||||
|
||||
s.controlMu.Lock()
|
||||
s.controlSeq++
|
||||
sequence := s.controlSeq
|
||||
delta := buildControlDeviceDelta(sequence, s.controlState, nextState)
|
||||
s.controlState = nextState
|
||||
subs := make([]*serverControlConn, 0, len(s.controlSubs))
|
||||
for _, sub := range s.controlSubs {
|
||||
subs = append(subs, sub)
|
||||
@@ -625,19 +678,255 @@ func (s *ServerService) broadcastChanged() {
|
||||
Sequence: sequence,
|
||||
}
|
||||
for _, sub := range subs {
|
||||
if supportsControlExtensions(sub.capabilities) {
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameDeviceDelta,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
}, delta, frame)
|
||||
continue
|
||||
}
|
||||
s.enqueueControlFrame(sub, frame)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlFrame(sub *serverControlConn, frame controlFrame) {
|
||||
s.enqueueControlMessage(sub, controlOutboundMessage{Frame: frame})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlPayload(sub *serverControlConn, frame controlFrame, payload any, fallback controlFrame) {
|
||||
rawPayload, err := marshalControlPayload(payload)
|
||||
if err != nil || len(rawPayload) > maxControlPayloadLength {
|
||||
s.enqueueControlFrame(sub, fallback)
|
||||
return
|
||||
}
|
||||
s.enqueueControlMessage(sub, controlOutboundMessage{Frame: frame, Payload: rawPayload})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlSnapshot(sub *serverControlConn, sequence uint64) {
|
||||
devices := s.buildDeviceStateV2()
|
||||
s.controlMu.Lock()
|
||||
s.controlState = deviceInfoV2Map(devices)
|
||||
s.controlMu.Unlock()
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameDeviceSnapshot,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
}, controlDeviceSnapshot{Sequence: sequence, Devices: devices}, controlFrame{
|
||||
Type: controlFrameChanged,
|
||||
Version: controlProtocolVersion,
|
||||
Sequence: sequence,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *ServerService) enqueueControlMessage(sub *serverControlConn, message controlOutboundMessage) {
|
||||
select {
|
||||
case sub.send <- frame:
|
||||
case sub.send <- message:
|
||||
default:
|
||||
s.logger.Debug("control subscriber ", sub.id, " lagged behind")
|
||||
_ = sub.conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) refreshControlState() {
|
||||
devices := s.buildDeviceStateV2()
|
||||
s.controlMu.Lock()
|
||||
s.controlState = deviceInfoV2Map(devices)
|
||||
s.controlMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 {
|
||||
busids := s.currentExports()
|
||||
if len(busids) == 0 {
|
||||
return nil
|
||||
}
|
||||
devices := make([]DeviceInfoV2, 0, len(busids))
|
||||
for _, busid := range busids {
|
||||
status, statusErr := s.ops.readUsbipStatus(busid)
|
||||
dev, devErr := s.ops.readSysfsDevice(busid, sysBusDevicePath(busid))
|
||||
if devErr != nil {
|
||||
devices = append(devices, DeviceInfoV2{
|
||||
BusID: busid,
|
||||
Backend: "linux-sysfs",
|
||||
StableID: "linux-busid:" + busid,
|
||||
State: deviceStateUnavailable,
|
||||
StatusReason: devErr.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
state := linuxUSBIPStatusState(status)
|
||||
reason := linuxUSBIPStatusReason(status)
|
||||
if statusErr != nil {
|
||||
state = deviceStateUnavailable
|
||||
reason = statusErr.Error()
|
||||
}
|
||||
entry := DeviceEntry{Info: dev.toProtocol(), Interfaces: dev.Interfaces}
|
||||
devices = append(devices, deviceInfoV2FromEntry(entry, "linux-sysfs", linuxStableID(dev), state, status, reason))
|
||||
}
|
||||
return devices
|
||||
}
|
||||
|
||||
func (s *ServerService) handleControlLeaseRequest(sub *serverControlConn, payload []byte) {
|
||||
var request controlLeaseRequest
|
||||
if err := unmarshalControlPayload(payload, &request); err != nil {
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameLeaseResponse,
|
||||
Version: controlProtocolVersion,
|
||||
}, controlLeaseResponse{
|
||||
ErrorCode: "bad_request",
|
||||
ErrorMessage: err.Error(),
|
||||
}, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()})
|
||||
return
|
||||
}
|
||||
response := s.createControlLeaseResponse(sub.id, request, s.linuxLeaseAvailable)
|
||||
s.enqueueControlPayload(sub, controlFrame{
|
||||
Type: controlFrameLeaseResponse,
|
||||
Version: controlProtocolVersion,
|
||||
}, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()})
|
||||
}
|
||||
|
||||
func (s *ServerService) linuxLeaseAvailable(busid string) (bool, string) {
|
||||
if !s.isExported(busid) {
|
||||
return false, "unknown busid"
|
||||
}
|
||||
status, err := s.ops.readUsbipStatus(busid)
|
||||
if err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
if status != usbipStatusAvailable {
|
||||
return false, linuxUSBIPStatusReason(status)
|
||||
}
|
||||
return true, ""
|
||||
}
|
||||
|
||||
func (s *ServerService) createControlLeaseResponse(subID uint64, request controlLeaseRequest, available func(string) (bool, string)) controlLeaseResponse {
|
||||
response := controlLeaseResponse{
|
||||
BusID: request.BusID,
|
||||
ClientNonce: request.ClientNonce,
|
||||
}
|
||||
if request.BusID == "" {
|
||||
response.ErrorCode = "bad_request"
|
||||
response.ErrorMessage = "missing busid"
|
||||
return response
|
||||
}
|
||||
if ok, reason := available(request.BusID); !ok {
|
||||
response.ErrorCode = "unavailable"
|
||||
response.ErrorMessage = reason
|
||||
return response
|
||||
}
|
||||
now := time.Now()
|
||||
expires := now.Add(importLeaseTTL)
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.cleanupExpiredImportLeasesLocked(now)
|
||||
if _, exists := s.leaseByBusID[request.BusID]; exists {
|
||||
response.ErrorCode = "busy"
|
||||
response.ErrorMessage = "lease already active"
|
||||
return response
|
||||
}
|
||||
s.leaseNextID++
|
||||
lease := serverImportLease{
|
||||
ID: s.leaseNextID,
|
||||
SubscriberID: subID,
|
||||
BusID: request.BusID,
|
||||
ClientNonce: request.ClientNonce,
|
||||
Generation: s.controlSeq,
|
||||
Expires: expires,
|
||||
}
|
||||
if s.leases == nil {
|
||||
s.leases = make(map[uint64]serverImportLease)
|
||||
}
|
||||
if s.leaseByBusID == nil {
|
||||
s.leaseByBusID = make(map[string]uint64)
|
||||
}
|
||||
s.leases[lease.ID] = lease
|
||||
s.leaseByBusID[lease.BusID] = lease.ID
|
||||
response.LeaseID = lease.ID
|
||||
response.Generation = lease.Generation
|
||||
response.TTLMillis = int64(importLeaseTTL / time.Millisecond)
|
||||
return response
|
||||
}
|
||||
|
||||
func (s *ServerService) consumeImportLease(request ImportExtRequest) bool {
|
||||
now := time.Now()
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
s.cleanupExpiredImportLeasesLocked(now)
|
||||
lease, ok := s.leases[request.LeaseID]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if lease.BusID != request.BusID || lease.ClientNonce != request.ClientNonce {
|
||||
return false
|
||||
}
|
||||
delete(s.leases, request.LeaseID)
|
||||
delete(s.leaseByBusID, request.BusID)
|
||||
return now.Before(lease.Expires)
|
||||
}
|
||||
|
||||
func (s *ServerService) cleanupExpiredImportLeasesLocked(now time.Time) {
|
||||
if s.leases == nil {
|
||||
s.leases = make(map[uint64]serverImportLease)
|
||||
}
|
||||
if s.leaseByBusID == nil {
|
||||
s.leaseByBusID = make(map[string]uint64)
|
||||
}
|
||||
for id, lease := range s.leases {
|
||||
if now.Before(lease.Expires) {
|
||||
continue
|
||||
}
|
||||
delete(s.leases, id)
|
||||
delete(s.leaseByBusID, lease.BusID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) deleteImportLeasesForSubscriberLocked(subID uint64) {
|
||||
for id, lease := range s.leases {
|
||||
if lease.SubscriberID != subID {
|
||||
continue
|
||||
}
|
||||
delete(s.leases, id)
|
||||
delete(s.leaseByBusID, lease.BusID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ServerService) currentControlSequence() uint64 {
|
||||
s.controlMu.Lock()
|
||||
defer s.controlMu.Unlock()
|
||||
return s.controlSeq
|
||||
}
|
||||
|
||||
func linuxStableID(d sysfsDevice) string {
|
||||
if d.Serial != "" {
|
||||
return "usb:" + hex16(d.VendorID) + ":" + hex16(d.ProductID) + ":" + d.Serial
|
||||
}
|
||||
return "linux-busid:" + d.BusID
|
||||
}
|
||||
|
||||
func linuxUSBIPStatusState(status int) string {
|
||||
switch status {
|
||||
case usbipStatusAvailable:
|
||||
return deviceStateAvailable
|
||||
case usbipStatusUsed:
|
||||
return deviceStateBusy
|
||||
default:
|
||||
return deviceStateUnavailable
|
||||
}
|
||||
}
|
||||
|
||||
func linuxUSBIPStatusReason(status int) string {
|
||||
switch status {
|
||||
case usbipStatusAvailable:
|
||||
return "available"
|
||||
case usbipStatusUsed:
|
||||
return "used"
|
||||
case usbipStatusError:
|
||||
return "error"
|
||||
default:
|
||||
return "status=" + hex32(uint32(status))
|
||||
}
|
||||
}
|
||||
|
||||
func sysBusDevicePath(busid string) string {
|
||||
return sysBusUSBDevices + "/" + busid
|
||||
}
|
||||
@@ -691,6 +980,10 @@ func hex16(v uint16) string {
|
||||
})
|
||||
}
|
||||
|
||||
func hex32(v uint32) string {
|
||||
return hex16(uint16(v>>16)) + hex16(uint16(v))
|
||||
}
|
||||
|
||||
func joinComma(parts []string) string {
|
||||
out := ""
|
||||
for i, p := range parts {
|
||||
|
||||
Reference in New Issue
Block a user