Merge f49c405509 into 9ffdf0c4a6

* chore(config): merge userConfig with defaultConfig and add a lock

* chore(cloud): add extra logging and tune timeout settings

cloud.go

@@ -24,6 +24,18 @@ type CloudRegisterRequest struct {
 	ClientId   string `json:"clientId"`
 }
 
+const (
+	// CloudWebSocketConnectTimeout is the timeout for the websocket connection to the cloud
+	CloudWebSocketConnectTimeout = 1 * time.Minute
+	// CloudAPIRequestTimeout is the timeout for cloud API requests
+	CloudAPIRequestTimeout = 10 * time.Second
+	// CloudOidcRequestTimeout is the timeout for OIDC token verification requests
+	// should be lower than the websocket response timeout set in cloud-api
+	CloudOidcRequestTimeout = 10 * time.Second
+	// CloudWebSocketPingInterval is the interval at which the websocket client sends ping messages to the cloud
+	CloudWebSocketPingInterval = 15 * time.Second
+)
+
 func handleCloudRegister(c *gin.Context) {
 	var req CloudRegisterRequest
 
@@ -44,22 +56,31 @@ func handleCloudRegister(c *gin.Context) {
 		return
 	}
 
-	resp, err := http.Post(req.CloudAPI+"/devices/token", "application/json", bytes.NewBuffer(jsonPayload))
+	client := &http.Client{Timeout: CloudAPIRequestTimeout}
+
+	apiReq, err := http.NewRequest(http.MethodPost, config.CloudURL+"/devices/token", bytes.NewBuffer(jsonPayload))
+	if err != nil {
+		c.JSON(500, gin.H{"error": "Failed to create register request: " + err.Error()})
+		return
+	}
+	apiReq.Header.Set("Content-Type", "application/json")
+
+	apiResp, err := client.Do(apiReq)
 	if err != nil {
 		c.JSON(500, gin.H{"error": "Failed to exchange token: " + err.Error()})
 		return
 	}
-	defer resp.Body.Close()
+	defer apiResp.Body.Close()
 
-	if resp.StatusCode != http.StatusOK {
+	if apiResp.StatusCode != http.StatusOK {
-		c.JSON(resp.StatusCode, gin.H{"error": "Failed to exchange token: " + resp.Status})
+		c.JSON(apiResp.StatusCode, gin.H{"error": "Failed to exchange token: " + apiResp.Status})
 		return
 	}
 
 	var tokenResp struct {
 		SecretToken string `json:"secretToken"`
 	}
-	if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil {
+	if err := json.NewDecoder(apiResp.Body).Decode(&tokenResp); err != nil {
 		c.JSON(500, gin.H{"error": "Failed to parse token response: " + err.Error()})
 		return
 	}
@@ -70,7 +91,7 @@ func handleCloudRegister(c *gin.Context) {
 	}
 
 	if config.CloudToken == "" {
-		logger.Info("Starting websocket client due to adoption")
+		cloudLogger.Info("Starting websocket client due to adoption")
 		go RunWebsocketClient()
 	}
 
@@ -122,7 +143,7 @@ func runWebsocketClient() error {
 	header := http.Header{}
 	header.Set("X-Device-ID", GetDeviceID())
 	header.Set("Authorization", "Bearer "+config.CloudToken)
-	dialCtx, cancelDial := context.WithTimeout(context.Background(), time.Minute)
+	dialCtx, cancelDial := context.WithTimeout(context.Background(), CloudWebSocketConnectTimeout)
 	defer cancelDial()
 	c, _, err := websocket.Dial(dialCtx, wsURL.String(), &websocket.DialOptions{
 		HTTPHeader: header,
@@ -131,15 +152,15 @@ func runWebsocketClient() error {
 		return err
 	}
 	defer c.CloseNow()
-	logger.Infof("WS connected to %v", wsURL.String())
+	cloudLogger.Infof("websocket connected to %s", wsURL.String())
 	runCtx, cancelRun := context.WithCancel(context.Background())
 	defer cancelRun()
 	go func() {
 		for {
-			time.Sleep(15 * time.Second)
+			time.Sleep(CloudWebSocketPingInterval)
 			err := c.Ping(runCtx)
 			if err != nil {
-				logger.Warnf("websocket ping error: %v", err)
+				cloudLogger.Warnf("websocket ping error: %v", err)
 				cancelRun()
 				return
 			}
@@ -157,24 +178,30 @@ func runWebsocketClient() error {
 		var req WebRTCSessionRequest
 		err = json.Unmarshal(msg, &req)
 		if err != nil {
-			logger.Warnf("unable to parse ws message: %v", string(msg))
+			cloudLogger.Warnf("unable to parse ws message: %v", string(msg))
 			continue
 		}
 
+		cloudLogger.Infof("new session request: %v", req.OidcGoogle)
+		cloudLogger.Tracef("session request info: %v", req)
+
 		err = handleSessionRequest(runCtx, c, req)
 		if err != nil {
-			logger.Infof("error starting new session: %v", err)
+			cloudLogger.Infof("error starting new session: %v", err)
 			continue
 		}
 	}
 }
 
 func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSessionRequest) error {
-	oidcCtx, cancelOIDC := context.WithTimeout(ctx, time.Minute)
+	oidcCtx, cancelOIDC := context.WithTimeout(ctx, CloudOidcRequestTimeout)
 	defer cancelOIDC()
 	provider, err := oidc.NewProvider(oidcCtx, "https://accounts.google.com")
 	if err != nil {
-		fmt.Println("Failed to initialize OIDC provider:", err)
+		_ = wsjson.Write(context.Background(), c, gin.H{
+			"error": fmt.Sprintf("failed to initialize OIDC provider: %v", err),
+		})
+		cloudLogger.Errorf("failed to initialize OIDC provider: %v", err)
 		return err
 	}
 
@@ -190,6 +217,7 @@ func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSess
 
 	googleIdentity := idToken.Audience[0] + ":" + idToken.Subject
 	if config.GoogleIdentity != googleIdentity {
+		_ = wsjson.Write(context.Background(), c, gin.H{"error": "google identity mismatch"})
 		return fmt.Errorf("google identity mismatch")
 	}
 
@@ -216,6 +244,9 @@ func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSess
 			_ = peerConn.Close()
 		}()
 	}
+
+	cloudLogger.Info("new session accepted")
+	cloudLogger.Tracef("new session accepted: %v", session)
 	currentSession = session
 	_ = wsjson.Write(context.Background(), c, gin.H{"sd": sd})
 	return nil
@@ -225,7 +256,7 @@ func RunWebsocketClient() {
 	for {
 		err := runWebsocketClient()
 		if err != nil {
-			fmt.Println("Websocket client error:", err)
+			cloudLogger.Errorf("websocket client error: %v", err)
 			time.Sleep(5 * time.Second)
 		}
 	}
@@ -254,7 +285,7 @@ func rpcDeregisterDevice() error {
 	}
 
 	req.Header.Set("Authorization", "Bearer "+config.CloudToken)
-	client := &http.Client{Timeout: 10 * time.Second}
+	client := &http.Client{Timeout: CloudAPIRequestTimeout}
 	resp, err := client.Do(req)
 	if err != nil {
 		return fmt.Errorf("failed to send deregister request: %w", err)

config.go

@@ -47,6 +47,9 @@ var (
 )
 
 func LoadConfig() {
+	configLock.Lock()
+	defer configLock.Unlock()
+
 	if config != nil {
 		logger.Info("config already loaded, skipping")
 		return

dev_deploy.sh

@@ -58,6 +58,9 @@ make build_dev
 # Change directory to the binary output directory
 cd bin
 
+# Kill any existing instances of the application
+ssh "${REMOTE_USER}@${REMOTE_HOST}" "killall jetkvm_app_debug || true"
+
 # Copy the binary to the remote host
 cat jetkvm_app | ssh "${REMOTE_USER}@${REMOTE_HOST}" "cat > $REMOTE_PATH/jetkvm_app_debug"
 
@@ -79,8 +82,7 @@ cd "$REMOTE_PATH"
 chmod +x jetkvm_app_debug
 
 # Run the application in the background
-./jetkvm_app_debug
+PION_LOG_TRACE=jetkvm,cloud ./jetkvm_app_debug
-
 EOF
 
 echo "Deployment complete."

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/pion/webrtc/v4 v4.0.0
 	github.com/pojntfx/go-nbd v0.3.2
 	github.com/psanford/httpreadat v0.1.0
+	github.com/ulikunitz/xz v0.5.12
 	github.com/vishvananda/netlink v1.3.0
 	go.bug.st/serial v1.6.2
 	golang.org/x/crypto v0.28.0

go.sum

@@ -142,6 +142,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/ulikunitz/xz v0.5.12 h1:37Nm15o69RwBkXM0J6A5OlE67RZTfzUxTj8fB3dfcsc=
+github.com/ulikunitz/xz v0.5.12/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=
 github.com/vishvananda/netlink v1.3.0 h1:X7l42GfcV4S6E4vHTsw48qbrV+9PVojNfIhZcwQdrZk=
 github.com/vishvananda/netlink v1.3.0/go.mod h1:i6NetklAujEcC6fK0JPjT8qSwWyO0HLn4UKG+hGqeJs=
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=

log.go

@@ -6,3 +6,4 @@ import "github.com/pion/logging"
 // ref: https://github.com/pion/webrtc/wiki/Debugging-WebRTC
 var logger = logging.NewDefaultLoggerFactory().NewLogger("jetkvm")
 var usbLogger = logging.NewDefaultLoggerFactory().NewLogger("usb")
+var cloudLogger = logging.NewDefaultLoggerFactory().NewLogger("cloud")

ui/dev_device.sh

@@ -1,21 +1,19 @@
 #!/bin/bash
 
+# Check if an IP address was provided as an argument
+if [ -z "$1" ]; then
+    echo "Usage: $0 <JetKVM IP Address>"
+    exit 1
+fi
+
+ip_address="$1"
+
 # Print header
 echo "┌──────────────────────────────────────┐"
 echo "│     JetKVM Development Setup         │"
 echo "└──────────────────────────────────────┘"
 
-# Prompt for IP address
-printf "Please enter the IP address of your JetKVM device: "
-read ip_address
-
-# Validate input is not empty
-if [ -z "$ip_address" ]; then
-    echo "Error: IP address cannot be empty"
-    exit 1
-fi
-
 # Set the environment variable and run Vite
 echo "Starting development server with JetKVM device at: $ip_address"
 sleep 1
-JETKVM_PROXY_URL="http://$ip_address" vite dev --mode=device
+JETKVM_PROXY_URL="http://$ip_address" npx vite dev --mode=device

usb_mass_storage.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/pion/webrtc/v4"
+	"github.com/ulikunitz/xz"
 )
 
 const massStorageName = "mass_storage.usb0"
@@ -269,16 +270,48 @@ func rpcMountWithStorage(filename string, mode VirtualMediaMode) error {
 		return fmt.Errorf("failed to get file info: %w", err)
 	}
 
-	err = setMassStorageImage(fullPath)
+	// Handle XZ compressed images
+	if strings.HasSuffix(filename, ".img.xz") {
+		logger.Info("Mounting compressed XZ image")
+
+		// Create temporary file for decompressed image
+		decompressedPath := filepath.Join(imagesFolder, strings.TrimSuffix(filename, ".xz")+".temp")
+		defer os.Remove(decompressedPath) // Clean up temp file after mounting
+
+		err = decompressXZImage(fullPath, decompressedPath)
+		if err != nil {
+			return fmt.Errorf("failed to decompress XZ image: %w", err)
+		}
+
+		// Mount the decompressed image
+		err = mountImage(decompressedPath)
+		if err != nil {
+			return fmt.Errorf("failed to mount decompressed image: %w", err)
+		}
+
+		currentVirtualMediaState = &VirtualMediaState{
+			Source:   Storage,
+			Mode:     mode,
+			Filename: filename,
+			Size:     fileInfo.Size(),
+		}
+
+		return nil
+	}
+
+	// Handle regular images
+	err = mountImage(fullPath)
 	if err != nil {
 		return fmt.Errorf("failed to set mass storage image: %w", err)
 	}
+
 	currentVirtualMediaState = &VirtualMediaState{
 		Source:   Storage,
 		Mode:     mode,
 		Filename: filename,
 		Size:     fileInfo.Size(),
 	}
+
 	return nil
 }
 
@@ -556,3 +589,30 @@ func handleUploadHttp(c *gin.Context) {
 
 	c.JSON(http.StatusOK, gin.H{"message": "Upload completed"})
 }
+
+// Add this helper function to handle XZ decompression
+func decompressXZImage(sourcePath, destPath string) error {
+	sourceFile, err := os.Open(sourcePath)
+	if err != nil {
+		return fmt.Errorf("failed to open source file: %w", err)
+	}
+	defer sourceFile.Close()
+
+	reader, err := xz.NewReader(sourceFile)
+	if err != nil {
+		return fmt.Errorf("failed to create XZ reader: %w", err)
+	}
+
+	destFile, err := os.Create(destPath)
+	if err != nil {
+		return fmt.Errorf("failed to create destination file: %w", err)
+	}
+	defer destFile.Close()
+
+	_, err = io.Copy(destFile, reader)
+	if err != nil {
+		return fmt.Errorf("failed to decompress file: %w", err)
+	}
+
+	return nil
+}