feat: add usbredir

Signed-off-by: Yaroslav Borbat <[email protected]>

Author: yaroslavborbat

api/client/kubeclient/async.go

@@ -127,10 +127,7 @@ func asyncSubresourceHelper(
 	case err = <-errChan:
 		return nil, err
 	case ws := <-aws.Connection:
-		return &wsStreamer{
-			conn: ws,
-			done: done,
-		}, nil
+		return newWebsocketStreamer(ws, done), nil
 	}
 }
 

api/client/kubeclient/streamer.go

@@ -75,7 +75,7 @@ func (c *wsConn) SetDeadline(t time.Time) error {
 	return c.Conn.SetReadDeadline(t)
 }
 
-func NewWebsocketStreamer(conn *websocket.Conn, done chan struct{}) *wsStreamer {
+func newWebsocketStreamer(conn *websocket.Conn, done chan struct{}) *wsStreamer {
 	return &wsStreamer{
 		conn: conn,
 		done: done,

src/cli/internal/cmd/usbredir/redirclient/client.go

@@ -0,0 +1,177 @@
+package redirclient
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"os/exec"
+	"time"
+
+	"github.com/deckhouse/virtualization/api/client/generated/clientset/versioned/typed/core/v1alpha2"
+)
+
+const usbRedirectClient string = "usbredirect"
+
+type ClientConnectFn func(ctx context.Context, device, address string) error
+
+type Client struct {
+	// To connect local USB device buffer to the remote VM using the websocket.
+	inputReader  *io.PipeReader
+	inputWriter  *io.PipeWriter
+	outputReader *io.PipeReader
+	outputWriter *io.PipeWriter
+
+	listener *net.TCPListener
+
+	// channels
+	done   chan struct{}
+	stream chan error
+	local  chan error
+	remote chan error
+
+	ctx context.Context
+
+	ClientConnect ClientConnectFn
+}
+
+func NewUSBRedirClient(ctx context.Context, address string, stream v1alpha2.StreamInterface) (*Client, error) {
+	inReader, inWriter := io.Pipe()
+	outReader, outWriter := io.Pipe()
+	k := &Client{
+		ctx:           ctx,
+		inputReader:   inReader,
+		inputWriter:   inWriter,
+		outputReader:  outReader,
+		outputWriter:  outWriter,
+		ClientConnect: clientConnect,
+	}
+
+	// Create local TCP server for usbredir client to connect
+	if err := k.withLocalTCPClient(address); err != nil {
+		return nil, err
+	}
+
+	// Start stream with remote usbredir endpoint
+	k.withRemoteVMIStream(stream)
+
+	// Connects data from local usbredir data to remote usbredir endpoint
+	k.proxyUSBRedir()
+
+	return k, nil
+}
+
+func (k *Client) withRemoteVMIStream(usbredirStream v1alpha2.StreamInterface) {
+	k.stream = make(chan error)
+
+	go func() {
+		defer k.outputWriter.Close()
+		select {
+		case k.stream <- usbredirStream.Stream(
+			v1alpha2.StreamOptions{
+				In:  k.inputReader,
+				Out: k.outputWriter,
+			},
+		):
+		case <-k.ctx.Done():
+		}
+	}()
+}
+
+func (k *Client) withLocalTCPClient(address string) error {
+	lnAddr, err := net.ResolveTCPAddr("tcp", address)
+	if err != nil {
+		return fmt.Errorf("Can't resolve the address: %s", err.Error())
+	}
+
+	// The local tcp server is used to proxy between remote websocket and local USB
+	k.listener, err = net.ListenTCP("tcp", lnAddr)
+	if err != nil {
+		return fmt.Errorf("Can't listen: %s", err.Error())
+	}
+
+	return nil
+}
+
+func (k *Client) proxyUSBRedir() {
+	// forward data to/from websocket after usbredir client connects.
+	k.done = make(chan struct{}, 1)
+	k.remote = make(chan error)
+	go func() {
+		defer k.inputWriter.Close()
+		start := time.Now()
+
+		usbredirConn, err := k.listener.Accept()
+		if err != nil {
+			fmt.Printf("Failed to accept connection: %s\n", err.Error())
+			k.remote <- err
+			return
+		}
+		defer usbredirConn.Close()
+
+		fmt.Printf("Connected to %s at %v\n", usbRedirectClient, time.Now().Sub(start))
+
+		stream := make(chan error)
+		// write to local usbredir from pipeOutReader
+		go func() {
+			_, err := io.Copy(usbredirConn, k.outputReader)
+			stream <- err
+		}()
+
+		// read from local usbredir towards pipeInWriter
+		go func() {
+			_, err := io.Copy(k.inputWriter, usbredirConn)
+			stream <- err
+		}()
+
+		select {
+		case <-k.done: // Wait for local usbredir to complete
+		case err = <-stream: // Wait for remote connection to close
+			if err == nil {
+				// Remote connection closed, report this as error
+				err = fmt.Errorf("Remote connection has closed.")
+			}
+			// Wait for local usbredir to complete
+			k.remote <- err
+		case <-k.ctx.Done():
+		}
+	}()
+}
+
+func clientConnect(ctx context.Context, device, address string) error {
+	bin := usbRedirectClient
+	args := []string{}
+	args = append(args, "--device", device, "--to", address)
+
+	fmt.Printf("port_arg: '%s'\n", address)
+	fmt.Printf("args: '%v'\n", args)
+	fmt.Printf("Executing commandline: '%s %v'\n", bin, args)
+
+	command := exec.CommandContext(ctx, bin, args...)
+	output, err := command.CombinedOutput()
+
+	fmt.Printf("%v output: %v", bin, string(output))
+
+	return err
+}
+
+func (k *Client) Redirect(device string) error {
+	// execute local usbredir binary
+	address := k.listener.Addr().String()
+	k.local = make(chan error)
+	go func() {
+		defer close(k.done)
+		k.local <- k.ClientConnect(k.ctx, device, address)
+	}()
+
+	var err error
+	select {
+	case err = <-k.stream:
+	case err = <-k.local:
+	case err = <-k.remote:
+	case <-k.ctx.Done():
+		err = k.ctx.Err()
+
+	}
+	return err
+}

src/cli/internal/cmd/usbredir/usbredir.go

@@ -24,6 +24,7 @@ import (
 	"github.com/spf13/pflag"
 
 	"github.com/deckhouse/virtualization/src/cli/internal/clientconfig"
+	"github.com/deckhouse/virtualization/src/cli/internal/cmd/usbredir/redirclient"
 	"github.com/deckhouse/virtualization/src/cli/internal/templates"
 )
 
@@ -105,8 +106,14 @@ func (c *USBRedir) Run(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	redir := NewClient(stream, newUsbRedirector(usbRedirectClient))
-	device := fmt.Sprintf("%s-%s", c.bus, c.device)
-
-	return redir.Redirect(cmd.Context(), device)
+	if usbRedirClient, err := redirclient.NewUSBRedirClient(cmd.Context(), "localhost:0", stream); err != nil {
+		return err
+	} else {
+		device := fmt.Sprintf("%s-%s", c.bus, c.device)
+		return usbRedirClient.Redirect(device)
+	}
+	//redir := NewClient(stream, newUsbRedirector(usbRedirectClient))
+	//device := fmt.Sprintf("%s-%s", c.bus, c.device)
+	//
+	//return redir.Redirect(cmd.Context(), device)
 }

templates/virtualization-api/rbac-for-us.yaml

@@ -72,6 +72,7 @@ rules:
   - virtualmachineinstances/unfreeze
   - virtualmachineinstances/addvolume
   - virtualmachineinstances/removevolume
+  - virtualmachineinstances/usbredir
   verbs:
   - get
   - patch
@@ -100,6 +101,7 @@ rules:
   - virtualmachines/removevolume
   - virtualmachines/unfreeze
   - virtualmachines/vnc
+  - virtualmachines/usbredir
   verbs:
   - get
   - patch