vnc proxy

Cargo.toml

@@ -51,6 +51,7 @@ tempfile = "3.10.1"
 test-common = { path = "test-common" }
 thouart = { git = "https://github.com/oxidecomputer/thouart.git" }
 tokio = { version = "1.36.0", features = ["full"] }
+tokio-tungstenite = "0.20.1"
 toml = "0.8.12"
 toml_edit = "0.22.9"
 url = "2.5.0"

cli/Cargo.toml

@@ -41,6 +41,7 @@ serde_json = { workspace = true }
 tabwriter = { workspace = true }
 thouart = { workspace = true }
 tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
 

cli/src/cli_builder.rs

@@ -274,6 +274,7 @@ fn xxx<'a>(command: CliCommand) -> Option<&'a str> {
         CliCommand::InstanceReboot => Some("instance reboot"),
         CliCommand::InstanceSerialConsole => None, // Special-cased
         CliCommand::InstanceSerialConsoleStream => None, // Ditto
+        CliCommand::InstanceVnc => None,           // Ditto
         CliCommand::InstanceStart => Some("instance start"),
         CliCommand::InstanceStop => Some("instance stop"),
         CliCommand::InstanceExternalIpList => Some("instance external-ip list"),

cli/src/cmd_instance.rs

@@ -8,14 +8,20 @@ use crate::RunnableCmd;
 use anyhow::Result;
 use async_trait::async_trait;
 use clap::Parser;
+use oxide::types::InstanceState;
 use oxide::types::{
     ByteCount, DiskSource, ExternalIpCreate, InstanceCpuCount, InstanceDiskAttachment, Name,
     NameOrId,
 };
 
+use futures::{SinkExt, StreamExt};
 use oxide::ClientImagesExt;
 use oxide::ClientInstancesExt;
 use std::path::PathBuf;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio_tungstenite::tungstenite::protocol::{frame::coding::CloseCode, CloseFrame, Role};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::WebSocketStream;
 
 /// Connect to or retrieve data from the instance's serial console.
 #[derive(Parser, Debug, Clone)]
@@ -165,7 +171,6 @@ pub struct CmdInstanceSerialHistory {
 
 #[async_trait]
 impl RunnableCmd for CmdInstanceSerialHistory {
-    // cli process becomes an interactive remote shell.
     async fn run(&self, ctx: &oxide::context::Context) -> Result<()> {
         let mut req = ctx
             .client()?
@@ -198,6 +203,177 @@ impl RunnableCmd for CmdInstanceSerialHistory {
     }
 }
 
+/// Connect to the instance's framebuffer and input with a local VNC client.
+#[derive(Parser, Debug, Clone)]
+#[command(verbatim_doc_comment)]
+#[command(name = "serial")]
+pub struct CmdInstanceVnc {
+    /// Name or ID of the instance
+    #[clap(long, short)]
+    instance: NameOrId,
+
+    /// Name or ID of the project
+    #[clap(long, short)]
+    project: Option<NameOrId>,
+    // TODO: vncviewer executable, or flag that says not to
+}
+
+#[async_trait]
+impl RunnableCmd for CmdInstanceVnc {
+    async fn run(&self, ctx: &oxide::context::Context) -> Result<()> {
+        let mut prereq = ctx
+            .client()?
+            .instance_view()
+            .instance(self.instance.clone());
+        let mut req = ctx.client()?.instance_vnc().instance(self.instance.clone());
+
+        if let Some(value) = &self.project {
+            req = req.project(value.clone());
+            prereq = prereq.project(value.clone());
+        } else if let NameOrId::Name(_) = &self.instance {
+            // on the server end, the connection is upgraded by the server
+            // before the worker thread attempts to look up the instance.
+            anyhow::bail!("Must provide --project when specifying instance by name rather than ID");
+        }
+
+        let view = prereq.send().await?;
+        if view.run_state != InstanceState::Running {
+            anyhow::bail!(
+                "Instance must be running to connect to VNC, but it is currently {:?}",
+                view.run_state
+            );
+        }
+
+        // TODO: custom listen address
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
+        let addr = listener.local_addr()?;
+        // yes, two ':' between IP and port. otherwise VNC adds 5900 to it!
+        let vncviewer_arg = format!("{ip}::{port}", ip = addr.ip(), port = addr.port());
+
+        // TODO: custom args etc.
+        let mut cmd = std::process::Command::new("vncviewer");
+        cmd.arg(&vncviewer_arg);
+        let child_res = cmd.spawn();
+        if child_res.is_err() {
+            eprintln!(
+                "Please connect a VNC client to {ip} on TCP port {port}.\nFor example: vncviewer {vncviewer_arg}",
+                ip = addr.ip(),
+                port = addr.port(),
+                vncviewer_arg = vncviewer_arg,
+            );
+        }
+
+        // TODO: clearer error case communication
+        let Ok((tcp_stream, _addr)) = listener.accept().await else {
+            anyhow::bail!("Failed to accept connection from local VNC client");
+        };
+
+        // okay, we have a local client, now actually start requesting I/O through nexus
+        let upgraded = req.send().await.map_err(|e| e.into_untyped())?.into_inner();
+
+        let ws = WebSocketStream::from_raw_socket(upgraded, Role::Client, None).await;
+
+        let (mut ws_sink, mut ws_stream) = ws.split();
+        let (mut tcp_reader, mut tcp_writer) = tcp_stream.into_split();
+        let (closed_tx, mut closed_rx) = tokio::sync::oneshot::channel::<()>();
+
+        let mut jh = tokio::spawn(async move {
+            // medium-sized websocket payload
+            let mut tcp_read_buf = vec![0u8; 65535];
+            loop {
+                tokio::select! {
+                    _ = &mut closed_rx => break,
+                    num_bytes_res = tcp_reader.read(&mut tcp_read_buf) => {
+                        match num_bytes_res {
+                            Ok(num_bytes) => {
+                                ws_sink
+                                    .send(Message::Binary(Vec::from(&tcp_read_buf[..num_bytes])))
+                                    .await?;
+                            }
+                            Err(e) => {
+                                ws_sink.send(Message::Close(None)).await?;
+                                anyhow::bail!("Local client disconnected: {}", e);
+                            }
+                        }
+                    }
+                }
+            }
+            Ok(ws_sink)
+        });
+
+        let mut close_frame = None;
+        let mut task_joined = false;
+        loop {
+            tokio::select! {
+                res = &mut jh => {
+                    if let Ok(Ok(mut ws_sink)) = res {
+                        // take() avoids borrow checker complaint about code that sends close
+                        // if we don't join the handle in the select (below loop)
+                        let _ = ws_sink.send(Message::Close(close_frame.take())).await.is_ok();
+                    }
+                    task_joined = true;
+                    break;
+                }
+                msg = ws_stream.next() => {
+                    match msg {
+                        Some(Ok(Message::Binary(data))) => {
+                            let mut start = 0;
+                            while start < data.len() {
+                                match tcp_writer.write(&data[start..]).await {
+                                    Ok(num_bytes) => {
+                                        start += num_bytes;
+                                    }
+                                    Err(e) => {
+                                        close_frame = Some(CloseFrame {
+                                            code: CloseCode::Error,
+                                            reason: e.to_string().into(),
+                                        });
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                        Some(Ok(Message::Close(Some(CloseFrame {code, reason})))) => {
+                            match code {
+                                CloseCode::Abnormal
+                                | CloseCode::Error
+                                | CloseCode::Extension
+                                | CloseCode::Invalid
+                                | CloseCode::Policy
+                                | CloseCode::Protocol
+                                | CloseCode::Size
+                                | CloseCode::Unsupported => {
+                                    anyhow::bail!("Server disconnected: {}", reason.to_string());
+                                }
+                                _ => break,
+                            }
+                        }
+                        Some(Ok(Message::Close(None))) => {
+                            eprintln!("Connection closed.");
+                            break;
+                        }
+                        None => {
+                            eprintln!("Connection lost.");
+                            break;
+                        }
+                        _ => continue,
+                    }
+                }
+            }
+        }
+
+        if !task_joined {
+            // let _: the connection may have already been dropped at this point.
+            let _ = closed_tx.send(()).is_ok();
+            if let Ok(Ok(mut ws_sink)) = jh.await {
+                let _ = ws_sink.send(Message::Close(close_frame)).await.is_ok();
+            }
+        }
+
+        Ok(())
+    }
+}
+
 /// Launch an instance from a disk image.
 #[derive(Parser, Debug, Clone)]
 #[command(verbatim_doc_comment)]

cli/src/generated_cli.rs

@@ -66,6 +66,7 @@ impl<T: CliConfig> Cli<T> {
             CliCommand::InstanceSshPublicKeyList => Self::cli_instance_ssh_public_key_list(),
             CliCommand::InstanceStart => Self::cli_instance_start(),
             CliCommand::InstanceStop => Self::cli_instance_stop(),
+            CliCommand::InstanceVnc => Self::cli_instance_vnc(),
             CliCommand::ProjectIpPoolList => Self::cli_project_ip_pool_list(),
             CliCommand::ProjectIpPoolView => Self::cli_project_ip_pool_view(),
             CliCommand::LoginLocal => Self::cli_login_local(),
@@ -1922,6 +1923,25 @@ impl<T: CliConfig> Cli<T> {
             .about("Stop instance")
     }
 
+    pub fn cli_instance_vnc() -> clap::Command {
+        clap::Command::new("")
+            .arg(
+                clap::Arg::new("instance")
+                    .long("instance")
+                    .value_parser(clap::value_parser!(types::NameOrId))
+                    .required(true)
+                    .help("Name or ID of the instance"),
+            )
+            .arg(
+                clap::Arg::new("project")
+                    .long("project")
+                    .value_parser(clap::value_parser!(types::NameOrId))
+                    .required(false)
+                    .help("Name or ID of the project"),
+            )
+            .about("Stream instance VNC framebuffer")
+    }
+
     pub fn cli_project_ip_pool_list() -> clap::Command {
         clap::Command::new("")
             .arg(
@@ -5467,6 +5487,7 @@ impl<T: CliConfig> Cli<T> {
             }
             CliCommand::InstanceStart => self.execute_instance_start(matches).await,
             CliCommand::InstanceStop => self.execute_instance_stop(matches).await,
+            CliCommand::InstanceVnc => self.execute_instance_vnc(matches).await,
             CliCommand::ProjectIpPoolList => self.execute_project_ip_pool_list(matches).await,
             CliCommand::ProjectIpPoolView => self.execute_project_ip_pool_view(matches).await,
             CliCommand::LoginLocal => self.execute_login_local(matches).await,
@@ -7414,6 +7435,28 @@ impl<T: CliConfig> Cli<T> {
         }
     }
 
+    pub async fn execute_instance_vnc(&self, matches: &clap::ArgMatches) -> anyhow::Result<()> {
+        let mut request = self.client.instance_vnc();
+        if let Some(value) = matches.get_one::<types::NameOrId>("instance") {
+            request = request.instance(value.clone());
+        }
+
+        if let Some(value) = matches.get_one::<types::NameOrId>("project") {
+            request = request.project(value.clone());
+        }
+
+        self.config.execute_instance_vnc(matches, &mut request)?;
+        let result = request.send().await;
+        match result {
+            Ok(r) => {
+                todo!()
+            }
+            Err(r) => {
+                todo!()
+            }
+        }
+    }
+
     pub async fn execute_project_ip_pool_list(
         &self,
         matches: &clap::ArgMatches,
@@ -11853,6 +11896,14 @@ pub trait CliConfig {
         Ok(())
     }
 
+    fn execute_instance_vnc(
+        &self,
+        matches: &clap::ArgMatches,
+        request: &mut builder::InstanceVnc,
+    ) -> anyhow::Result<()> {
+        Ok(())
+    }
+
     fn execute_project_ip_pool_list(
         &self,
         matches: &clap::ArgMatches,
@@ -12925,6 +12976,7 @@ pub enum CliCommand {
     InstanceSshPublicKeyList,
     InstanceStart,
     InstanceStop,
+    InstanceVnc,
     ProjectIpPoolList,
     ProjectIpPoolView,
     LoginLocal,
@@ -13110,6 +13162,7 @@ impl CliCommand {
             CliCommand::InstanceSshPublicKeyList,
             CliCommand::InstanceStart,
             CliCommand::InstanceStop,
+            CliCommand::InstanceVnc,
             CliCommand::ProjectIpPoolList,
             CliCommand::ProjectIpPoolView,
             CliCommand::LoginLocal,

cli/src/main.rs

@@ -47,6 +47,7 @@ pub fn make_cli() -> NewCli<'static> {
         .add_custom::<cmd_version::CmdVersion>("version")
         .add_custom::<cmd_disk::CmdDiskImport>("disk import")
         .add_custom::<cmd_instance::CmdInstanceSerial>("instance serial")
+        .add_custom::<cmd_instance::CmdInstanceVnc>("instance vnc")
         .add_custom::<cmd_instance::CmdInstanceFromImage>("instance from-image")
         .add_custom::<cmd_completion::CmdCompletion>("completion")
 }

oxide.json

@@ -2676,6 +2676,45 @@
         }
       }
     },
+    "/v1/instances/{instance}/vnc": {
+      "get": {
+        "tags": [
+          "instances"
+        ],
+        "summary": "Stream instance VNC framebuffer",
+        "operationId": "instance_vnc",
+        "parameters": [
+          {
+            "in": "path",
+            "name": "instance",
+            "description": "Name or ID of the instance",
+            "required": true,
+            "schema": {
+              "$ref": "#/components/schemas/NameOrId"
+            }
+          },
+          {
+            "in": "query",
+            "name": "project",
+            "description": "Name or ID of the project",
+            "schema": {
+              "$ref": "#/components/schemas/NameOrId"
+            }
+          }
+        ],
+        "responses": {
+          "default": {
+            "description": "",
+            "content": {
+              "*/*": {
+                "schema": {}
+              }
+            }
+          }
+        },
+        "x-dropshot-websocket": {}
+      }
+    },
     "/v1/ip-pools": {
       "get": {
         "tags": [