hamlab/internal/incus/vga.go

package incus

// The graphical console. Incus exposes a VM's SPICE display as a websocket: POST to the
// instance's console endpoint with type vga and it answers an operation with two secrets,
// one for the display stream and one for a control channel. Ham Lab dials both over Incus's
// unix socket and hands the display stream to the browser, where spice-html5 draws it.

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/coder/websocket"
)

// SocketPath is where Incus's API listens; the daemon's account reaches it by being in
// incus-admin, which Install arranges.
func SocketPath() string {
	if p := os.Getenv("INCUS_SOCKET"); p != "" {
		return p
	}
	for _, p := range []string{"/var/lib/incus/unix.socket", "/run/incus/unix.socket"} {
		if _, err := os.Stat(p); err == nil {
			return p
		}
	}
	return "/var/lib/incus/unix.socket"
}

func (m *Manager) apiClient() *http.Client {
	sock := SocketPath()
	return &http.Client{Transport: &http.Transport{
		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "unix", sock)
		},
		ForceAttemptHTTP2: false,
	}}
}

// A SPICE client opens one connection per channel — main, display, inputs, cursor, playback
// — and spice-html5 does the same with websockets, all to one URL. Incus allows exactly one
// console operation per VM at a time, but takes any number of data websockets on it. So a VGA
// session is one operation, held for as long as any channel is open, and every browser
// websocket becomes one more data websocket on it.
type vgaSession struct {
	op      string
	secret  string
	control *websocket.Conn
	cancel  context.CancelFunc
	refs    int
}

// VGAChannel returns a fresh data websocket on the VM's console operation, opening the
// operation first when none is held. release must be called when the channel is done.
func (m *Manager) VGAChannel(ctx context.Context, name string) (data *websocket.Conn, release func(), err error) {
	if !ValidName(name) {
		return nil, nil, errors.New("bad name")
	}
	client := m.apiClient()
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.vga == nil {
		m.vga = map[string]*vgaSession{}
	}
	dial := func(s *vgaSession) (*websocket.Conn, error) {
		c, cancel := context.WithTimeout(ctx, 20*time.Second)
		defer cancel()
		u := fmt.Sprintf("ws://incus/1.0/operations/%s/websocket?secret=%s", s.op, s.secret)
		ws, _, err := websocket.Dial(c, u, &websocket.DialOptions{HTTPClient: client, CompressionMode: websocket.CompressionDisabled})
		return ws, err
	}
	s := m.vga[name]
	if s != nil {
		if ws, err := dial(s); err == nil {
			s.refs++
			ws.SetReadLimit(64 << 20)
			return ws, m.releaser(name, s), nil
		}
		// the operation is gone (the VM restarted, or Incus did): start over
		s.cancel()
		delete(m.vga, name)
	}
	s, err = m.openVGA(ctx, client, name, false)
	if err != nil && strings.Contains(err.Error(), "already connected") {
		// a session a previous Ham Lab process held and never let go of; take it over
		s, err = m.openVGA(ctx, client, name, true)
	}
	if err != nil {
		return nil, nil, err
	}
	ws, err := dial(s)
	if err != nil {
		s.cancel()
		return nil, nil, err
	}
	ws.SetReadLimit(64 << 20)
	s.refs = 1
	m.vga[name] = s
	return ws, m.releaser(name, s), nil
}

func (m *Manager) releaser(name string, s *vgaSession) func() {
	return func() {
		m.mu.Lock()
		defer m.mu.Unlock()
		s.refs--
		if s.refs <= 0 && m.vga[name] == s {
			s.cancel()
			delete(m.vga, name)
		}
	}
}

// openVGA asks Incus for the console operation and holds its control websocket, which is
// what keeps the operation alive.
func (m *Manager) openVGA(ctx context.Context, client *http.Client, name string, force bool) (*vgaSession, error) {
	body, _ := json.Marshal(map[string]any{"type": "vga", "width": 0, "height": 0, "force": force})
	c, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
	req, _ := http.NewRequestWithContext(c, "POST", "http://incus/1.0/instances/"+name+"/console", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	resp, err := client.Do(req)
	if err != nil {
		if isPermission(err) {
			return nil, errors.New("Ham Lab's account can't reach Incus's socket; Finish setup on the Virtualization tab puts it in incus-admin")
		}
		return nil, err
	}
	defer resp.Body.Close()
	var op struct {
		Error    string `json:"error"`
		Metadata struct {
			ID       string `json:"id"`
			Metadata struct {
				FDs map[string]string `json:"fds"`
			} `json:"metadata"`
		} `json:"metadata"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&op); err != nil {
		return nil, errors.New("unreadable answer from incus")
	}
	if op.Error != "" {
		return nil, errors.New(op.Error)
	}
	secret, controlSecret := op.Metadata.Metadata.FDs["0"], op.Metadata.Metadata.FDs["control"]
	if op.Metadata.ID == "" || secret == "" {
		return nil, errors.New("incus gave no console; is the VM running?")
	}
	s := &vgaSession{op: op.Metadata.ID, secret: secret}
	// the control websocket lives outside any one request: it is the session
	sctx, scancel := context.WithCancel(context.Background())
	s.cancel = scancel
	if controlSecret != "" {
		u := fmt.Sprintf("ws://incus/1.0/operations/%s/websocket?secret=%s", s.op, controlSecret)
		control, _, err := websocket.Dial(sctx, u, &websocket.DialOptions{HTTPClient: client, CompressionMode: websocket.CompressionDisabled})
		if err != nil {
			scancel()
			return nil, err
		}
		s.control = control
		go func() { // drain it; when it ends, so has the operation
			for {
				if _, _, err := control.Read(sctx); err != nil {
					break
				}
			}
			control.Close(websocket.StatusNormalClosure, "")
		}()
		go func() { <-sctx.Done(); control.Close(websocket.StatusNormalClosure, "") }()
	}
	return s, nil
}

func isPermission(err error) bool {
	var pe *os.PathError
	if errors.As(err, &pe) {
		return os.IsPermission(pe)
	}
	var oe *net.OpError
	if errors.As(err, &oe) {
		return os.IsPermission(oe.Err) || isPermission(oe.Err)
	}
	return false
}