Remote command execution in microVM instances via vsock

.github/workflows/test.yml

@@ -30,9 +30,9 @@ jobs:
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_PASSWORD }}
-
-      - name: Run tests
-        run: make test
-
+
       - name: Build
         run: make build
+
+      - name: Run tests
+        run: make test

.gitignore

@@ -17,3 +17,6 @@ tmp/**
 
 # Cloud Hypervisor binaries (embedded at build time)
 lib/vmm/binaries/cloud-hypervisor/*/*/cloud-hypervisor
+cloud-hypervisor
+cloud-hypervisor/**
+lib/system/exec_agent/exec-agent

Makefile

@@ -76,8 +76,15 @@ generate-wire: $(WIRE)
 	@echo "Generating wire code..."
 	cd ./cmd/api && $(WIRE)
 
+# Generate gRPC code from proto
+generate-grpc:
+	@echo "Generating gRPC code from proto..."
+	protoc --go_out=. --go_opt=paths=source_relative \
+		--go-grpc_out=. --go-grpc_opt=paths=source_relative \
+		lib/exec/exec.proto
+
 # Generate all code
-generate-all: oapi-generate generate-vmm-client generate-wire
+generate-all: oapi-generate generate-vmm-client generate-wire generate-grpc
 
 # Check if binaries exist, download if missing
 .PHONY: ensure-ch-binaries
@@ -87,16 +94,28 @@ ensure-ch-binaries:
 		$(MAKE) download-ch-binaries; \
 	fi
 
+# Build exec-agent (guest binary) into its own directory for embedding
+lib/system/exec_agent/exec-agent: lib/system/exec_agent/main.go
+	@echo "Building exec-agent..."
+	cd lib/system/exec_agent && CGO_ENABLED=0 go build -ldflags="-s -w" -o exec-agent .
+
 # Build the binary
-build: ensure-ch-binaries | $(BIN_DIR)
+build: ensure-ch-binaries lib/system/exec_agent/exec-agent | $(BIN_DIR)
 	go build -tags containers_image_openpgp -o $(BIN_DIR)/hypeman ./cmd/api
 
+# Build exec CLI
+build-exec: | $(BIN_DIR)
+	go build -o $(BIN_DIR)/hypeman-exec ./cmd/exec
+
+# Build all binaries
+build-all: build build-exec
+
 # Run in development mode with hot reload
 dev: $(AIR)
 	$(AIR) -c .air.toml
 
 # Run tests
-test: ensure-ch-binaries
+test: ensure-ch-binaries lib/system/exec_agent/exec-agent
 	go test -tags containers_image_openpgp -v -timeout 30s ./...
 
 # Generate JWT token for testing
@@ -109,4 +128,7 @@ clean:
 	rm -rf $(BIN_DIR)
 	rm -f lib/oapi/oapi.go
 	rm -f lib/vmm/vmm.go
+	rm -f lib/exec/exec.pb.go
+	rm -f lib/exec/exec_grpc.pb.go
+	rm -f lib/system/exec_agent/exec-agent
 

cmd/api/api/exec.go

@@ -0,0 +1,202 @@
+package api
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/gorilla/websocket"
+	"github.com/onkernel/hypeman/lib/exec"
+	"github.com/onkernel/hypeman/lib/instances"
+	"github.com/onkernel/hypeman/lib/logger"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  32 * 1024,
+	WriteBufferSize: 32 * 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		// Allow all origins for now - can be tightened in production
+		return true
+	},
+}
+
+// ExecRequest represents the JSON body for exec requests
+type ExecRequest struct {
+	Command []string          `json:"command"`
+	TTY     bool              `json:"tty"`
+	Env     map[string]string `json:"env,omitempty"`
+	Cwd     string            `json:"cwd,omitempty"`
+	Timeout int32             `json:"timeout,omitempty"` // seconds
+}
+
+// ExecHandler handles exec requests via WebSocket for bidirectional streaming
+func (s *ApiService) ExecHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	log := logger.FromContext(ctx)
+	startTime := time.Now()
+
+	instanceID := chi.URLParam(r, "id")
+
+	// Get instance
+	inst, err := s.InstanceManager.GetInstance(ctx, instanceID)
+	if err != nil {
+		if err == instances.ErrNotFound {
+			http.Error(w, `{"code":"not_found","message":"instance not found"}`, http.StatusNotFound)
+			return
+		}
+		log.ErrorContext(ctx, "failed to get instance", "error", err)
+		http.Error(w, `{"code":"internal_error","message":"failed to get instance"}`, http.StatusInternalServerError)
+		return
+	}
+
+	if inst.State != instances.StateRunning {
+		http.Error(w, fmt.Sprintf(`{"code":"invalid_state","message":"instance must be running (current state: %s)"}`, inst.State), http.StatusConflict)
+		return
+	}
+
+	// Upgrade to WebSocket first
+	ws, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.ErrorContext(ctx, "websocket upgrade failed", "error", err)
+		return
+	}
+	defer ws.Close()
+
+	// Read JSON request from first WebSocket message
+	msgType, message, err := ws.ReadMessage()
+	if err != nil {
+		log.ErrorContext(ctx, "failed to read exec request", "error", err)
+		ws.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"error":"failed to read request: %v"}`, err)))
+		return
+	}
+
+	if msgType != websocket.TextMessage {
+		log.ErrorContext(ctx, "expected text message with JSON request", "type", msgType)
+		ws.WriteMessage(websocket.TextMessage, []byte(`{"error":"first message must be JSON text"}`))
+		return
+	}
+
+	// Parse JSON request
+	var execReq ExecRequest
+	if err := json.Unmarshal(message, &execReq); err != nil {
+		log.ErrorContext(ctx, "invalid JSON request", "error", err)
+		ws.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"error":"invalid JSON: %v"}`, err)))
+		return
+	}
+
+	// Default command if not specified
+	if len(execReq.Command) == 0 {
+		execReq.Command = []string{"/bin/sh"}
+	}
+
+	// Get JWT subject for audit logging (if available)
+	subject := "unknown"
+	if claims, ok := r.Context().Value("claims").(map[string]interface{}); ok {
+		if sub, ok := claims["sub"].(string); ok {
+			subject = sub
+		}
+	}
+
+	// Audit log: exec session started
+	log.InfoContext(ctx, "exec session started",
+		"instance_id", instanceID,
+		"subject", subject,
+		"command", execReq.Command,
+		"tty", execReq.TTY,
+		"cwd", execReq.Cwd,
+		"timeout", execReq.Timeout,
+	)
+
+	// Create WebSocket read/writer wrapper
+	wsConn := &wsReadWriter{ws: ws, ctx: ctx}
+
+	// Execute via vsock
+	exit, err := exec.ExecIntoInstance(ctx, inst.VsockSocket, exec.ExecOptions{
+		Command: execReq.Command,
+		Stdin:   wsConn,
+		Stdout:  wsConn,
+		Stderr:  wsConn,
+		TTY:     execReq.TTY,
+		Env:     execReq.Env,
+		Cwd:     execReq.Cwd,
+		Timeout: execReq.Timeout,
+	})
+
+	duration := time.Since(startTime)
+
+	if err != nil {
+		log.ErrorContext(ctx, "exec failed",
+			"error", err,
+			"instance_id", instanceID,
+			"subject", subject,
+			"duration_ms", duration.Milliseconds(),
+		)
+		// Send error message over WebSocket before closing
+		ws.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf("Error: %v", err)))
+		return
+	}
+
+	// Audit log: exec session ended
+	log.InfoContext(ctx, "exec session ended",
+		"instance_id", instanceID,
+		"subject", subject,
+		"exit_code", exit.Code,
+		"duration_ms", duration.Milliseconds(),
+	)
+
+	// Send close frame with exit code in JSON
+	closeMsg := fmt.Sprintf(`{"exitCode":%d}`, exit.Code)
+	ws.WriteMessage(websocket.TextMessage, []byte(closeMsg))
+}
+
+// wsReadWriter wraps a WebSocket connection to implement io.ReadWriter
+type wsReadWriter struct {
+	ws     *websocket.Conn
+	ctx    context.Context
+	reader io.Reader
+	mu     sync.Mutex
+}
+
+func (w *wsReadWriter) Read(p []byte) (n int, err error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	// If we have a pending reader, continue reading from it
+	if w.reader != nil {
+		n, err = w.reader.Read(p)
+		if err != io.EOF {
+			return n, err
+		}
+		// EOF means we finished this message, get next one
+		w.reader = nil
+	}
+
+	// Read next WebSocket message
+	messageType, data, err := w.ws.ReadMessage()
+	if err != nil {
+		return 0, err
+	}
+
+	// Only handle binary and text messages
+	if messageType != websocket.BinaryMessage && messageType != websocket.TextMessage {
+		return 0, fmt.Errorf("unexpected message type: %d", messageType)
+	}
+
+	// Create reader for this message
+	w.reader = bytes.NewReader(data)
+	return w.reader.Read(p)
+}
+
+func (w *wsReadWriter) Write(p []byte) (n int, err error) {
+	if err := w.ws.WriteMessage(websocket.BinaryMessage, p); err != nil {
+		return 0, err
+	}
+	return len(p), nil
+}
+