Merge 3d770e73c8 into a42384fed6

block_device.go

@@ -25,15 +25,22 @@ func (r remoteImageBackend) ReadAt(p []byte, off int64) (n int, err error) {
 	mountedImageSize := currentVirtualMediaState.Size
 	virtualMediaStateMutex.RUnlock()
 
-	_, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
 	readLen := int64(len(p))
 	if off+readLen > mountedImageSize {
 		readLen = mountedImageSize - off
 	}
-
+	var data []byte
 	switch source {
+	case WebRTC:
+		data, err = webRTCDiskReader.Read(ctx, off, readLen)
+		if err != nil {
+			return 0, err
+		}
+		n = copy(p, data)
+		return n, nil
 	case HTTP:
 		return httpRangeReader.ReadAt(p, off)
 	default:

fuse.go

@@ -0,0 +1,114 @@
+package kvm
+
+import (
+	"context"
+	"os"
+	"sync"
+	"syscall"
+
+	"github.com/hanwen/go-fuse/v2/fs"
+	"github.com/hanwen/go-fuse/v2/fuse"
+)
+
+type WebRTCStreamFile struct {
+	fs.Inode
+	mu   sync.Mutex
+	Attr fuse.Attr
+	size uint64
+}
+
+var _ = (fs.NodeOpener)((*WebRTCStreamFile)(nil))
+var _ = (fs.NodeOpener)((*WebRTCStreamFile)(nil))
+var _ = (fs.NodeOpener)((*WebRTCStreamFile)(nil))
+var _ = (fs.NodeOpener)((*WebRTCStreamFile)(nil))
+var _ = (fs.NodeOpener)((*WebRTCStreamFile)(nil))
+
+func (f *WebRTCStreamFile) Open(ctx context.Context, flags uint32) (fh fs.FileHandle, fuseFlags uint32, errno syscall.Errno) {
+	return nil, fuse.FOPEN_KEEP_CACHE, fs.OK
+}
+
+func (f *WebRTCStreamFile) Write(ctx context.Context, fh fs.FileHandle, data []byte, off int64) (uint32, syscall.Errno) {
+	return 0, syscall.EROFS
+}
+
+var _ = (fs.NodeGetattrer)((*WebRTCStreamFile)(nil))
+
+func (f *WebRTCStreamFile) Getattr(ctx context.Context, fh fs.FileHandle, out *fuse.AttrOut) syscall.Errno {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out.Attr = f.Attr
+	out.Size = f.size
+	return fs.OK
+}
+
+func (f *WebRTCStreamFile) Setattr(ctx context.Context, fh fs.FileHandle, in *fuse.SetAttrIn, out *fuse.AttrOut) syscall.Errno {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out.Attr = f.Attr
+	return fs.OK
+}
+
+func (f *WebRTCStreamFile) Flush(ctx context.Context, fh fs.FileHandle) syscall.Errno {
+	return fs.OK
+}
+
+type DiskReadRequest struct {
+	Start uint64 `json:"start"`
+	End   uint64 `json:"end"`
+}
+
+var diskReadChan = make(chan []byte, 1)
+
+func (f *WebRTCStreamFile) Read(ctx context.Context, fh fs.FileHandle, dest []byte, off int64) (fuse.ReadResult, syscall.Errno) {
+	buf, err := webRTCDiskReader.Read(ctx, off, int64(len(dest)))
+	if err != nil {
+		return nil, syscall.EIO
+	}
+	return fuse.ReadResultData(buf), fs.OK
+}
+
+func (f *WebRTCStreamFile) SetSize(size uint64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.size = size
+}
+
+type FuseRoot struct {
+	fs.Inode
+}
+
+var webRTCStreamFile = &WebRTCStreamFile{}
+
+func (r *FuseRoot) OnAdd(ctx context.Context) {
+	ch := r.NewPersistentInode(ctx, webRTCStreamFile, fs.StableAttr{Ino: 2})
+	r.AddChild("disk", ch, false)
+}
+
+func (r *FuseRoot) Getattr(ctx context.Context, fh fs.FileHandle, out *fuse.AttrOut) syscall.Errno {
+	out.Mode = 0755
+	return 0
+}
+
+var _ = (fs.NodeGetattrer)((*FuseRoot)(nil))
+var _ = (fs.NodeOnAdder)((*FuseRoot)(nil))
+
+const fuseMountPoint = "/mnt/webrtc"
+
+var fuseServer *fuse.Server
+
+func RunFuseServer() {
+	opts := &fs.Options{}
+	opts.DirectMountStrict = true
+	_ = os.Mkdir(fuseMountPoint, 0755)
+	var err error
+	fuseServer, err = fs.Mount(fuseMountPoint, &FuseRoot{}, opts)
+	if err != nil {
+		logger.Warn().Err(err).Msg("failed to mount fuse")
+	}
+	fuseServer.Wait()
+}
+
+type WebRTCImage struct {
+	Size     uint64 `json:"size"`
+	Filename string `json:"filename"`
+}

jsonrpc.go

@@ -1103,6 +1103,7 @@ var rpcHandlers = map[string]RPCHandler{
 	"getVirtualMediaState":   {Func: rpcGetVirtualMediaState},
 	"getStorageSpace":        {Func: rpcGetStorageSpace},
 	"mountWithHTTP":          {Func: rpcMountWithHTTP, Params: []string{"url", "mode"}},
+	"mountWithWebRTC":        {Func: rpcMountWithWebRTC, Params: []string{"filename", "size", "mode"}},
 	"mountWithStorage":       {Func: rpcMountWithStorage, Params: []string{"filename", "mode"}},
 	"listStorageFiles":       {Func: rpcListStorageFiles},
 	"deleteStorageFile":      {Func: rpcDeleteStorageFile, Params: []string{"filename"}},

remote_mount.go

@@ -0,0 +1,62 @@
+package kvm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+)
+
+type RemoteImageReader interface {
+	Read(ctx context.Context, offset int64, size int64) ([]byte, error)
+}
+
+type WebRTCDiskReader struct {
+}
+
+var webRTCDiskReader WebRTCDiskReader
+
+func (w *WebRTCDiskReader) Read(ctx context.Context, offset int64, size int64) ([]byte, error) {
+	virtualMediaStateMutex.RLock()
+	if currentVirtualMediaState == nil {
+		virtualMediaStateMutex.RUnlock()
+		return nil, errors.New("image not mounted")
+	}
+	if currentVirtualMediaState.Source != WebRTC {
+		virtualMediaStateMutex.RUnlock()
+		return nil, errors.New("image not mounted from webrtc")
+	}
+	mountedImageSize := currentVirtualMediaState.Size
+	virtualMediaStateMutex.RUnlock()
+	end := min(offset+size, mountedImageSize)
+	req := DiskReadRequest{
+		Start: uint64(offset),
+		End:   uint64(end),
+	}
+	jsonBytes, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if currentSession == nil || currentSession.DiskChannel == nil {
+		return nil, errors.New("not active session")
+	}
+
+	logger.Debug().Str("request", string(jsonBytes)).Msg("reading from webrtc")
+	err = currentSession.DiskChannel.SendText(string(jsonBytes))
+	if err != nil {
+		return nil, err
+	}
+	var buf []byte
+	for {
+		select {
+		case data := <-diskReadChan:
+			buf = data[16:]
+		case <-ctx.Done():
+			return nil, context.Canceled
+		}
+		if len(buf) >= int(end-offset) {
+			break
+		}
+	}
+	return buf, nil
+}

usb_mass_storage.go

@@ -69,6 +69,11 @@ func setMassStorageMode(cdrom bool) error {
 	return gadget.UpdateGadgetConfig()
 }
 
+func onDiskMessage(msg webrtc.DataChannelMessage) {
+	logger.Info().Int("len", len(msg.Data)).Msg("Disk Message")
+	diskReadChan <- msg.Data
+}
+
 func mountImage(imagePath string) error {
 	err := setMassStorageImage("")
 	if err != nil {
@@ -229,6 +234,7 @@ func getInitialVirtualMediaState() (*VirtualMediaState, error) {
 		initialState.Mode = CDROM
 	}
 
+	// TODO: check if it's WebRTC or HTTP
 	switch diskPath {
 	case "":
 		return nil, nil
@@ -307,6 +313,43 @@ func rpcMountWithHTTP(url string, mode VirtualMediaMode) error {
 	return nil
 }
 
+func rpcMountWithWebRTC(filename string, size int64, mode VirtualMediaMode) error {
+	virtualMediaStateMutex.Lock()
+	if currentVirtualMediaState != nil {
+		virtualMediaStateMutex.Unlock()
+		return fmt.Errorf("another virtual media is already mounted")
+	}
+	currentVirtualMediaState = &VirtualMediaState{
+		Source:   WebRTC,
+		Mode:     mode,
+		Filename: filename,
+		Size:     size,
+	}
+	virtualMediaStateMutex.Unlock()
+
+	if err := setMassStorageMode(mode == CDROM); err != nil {
+		return fmt.Errorf("failed to set mass storage mode: %w", err)
+	}
+
+	logger.Debug().Interface("currentVirtualMediaState", currentVirtualMediaState).Msg("currentVirtualMediaState")
+	logger.Debug().Msg("Starting nbd device")
+	nbdDevice = NewNBDDevice()
+	err := nbdDevice.Start()
+	if err != nil {
+		logger.Warn().Err(err).Msg("failed to start nbd device")
+		return err
+	}
+	logger.Debug().Msg("nbd device started")
+	//TODO: replace by polling on block device having right size
+	time.Sleep(1 * time.Second)
+	err = setMassStorageImage("/dev/nbd0")
+	if err != nil {
+		return err
+	}
+	logger.Info().Msg("usb mass storage mounted")
+	return nil
+}
+
 func rpcMountWithStorage(filename string, mode VirtualMediaMode) error {
 	filename, err := sanitizeFilename(filename)
 	if err != nil {

webrtc.go

@@ -126,6 +126,9 @@ func newSession(config SessionConfig) (*Session, error) {
 			triggerOTAStateUpdate()
 			triggerVideoStateUpdate()
 			triggerUSBStateUpdate()
+		case "disk":
+			session.DiskChannel = d
+			d.OnMessage(onDiskMessage)
 		case "terminal":
 			handleTerminalChannel(d)
 		case "serial":