1use std::net::SocketAddr;
13use std::path::{Path, PathBuf};
14use std::process::Stdio;
15use std::sync::Arc;
16use std::time::Duration;
17
18use axum::body::{Body, to_bytes};
19use axum::extract::ws::rejection::WebSocketUpgradeRejection;
20use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
21use axum::http::{HeaderMap, StatusCode, Uri, header};
22use axum::response::{IntoResponse, Response};
23use futures::{SinkExt, StreamExt};
24#[cfg(unix)]
29use tokio::io::AsyncBufReadExt;
30#[cfg(unix)]
31use tokio::net::UnixStream;
32use tokio::process::Child;
33#[cfg(unix)]
34use tokio::process::Command;
35use tokio::sync::Mutex;
36use tokio::task::JoinHandle;
37use tokio_tungstenite::tungstenite;
38
39use crate::host::bootstrap::TakeoverProviderKind;
40use crate::host::browser::{pick_ephemeral_port, resolve_named_bin, wait_for_tcp_ready};
41use crate::shared::error::{Error, ErrorCode};
42
43pub const DISPLAY_WIDTH: u16 = 1280;
47pub const DISPLAY_HEIGHT: u16 = 720;
48
49pub struct ProviderHandle {
52 pub display: String,
53 pub web_port: u16,
54 pub window_manager: bool,
59 _rfb_port: u16,
60 child: Mutex<Child>,
61 wm_child: Option<Mutex<Child>>,
62 stderr_task: Option<JoinHandle<()>>,
63}
64
65impl Drop for ProviderHandle {
66 fn drop(&mut self) {
67 if let Some(task) = self.stderr_task.take() {
68 task.abort();
69 }
70 if let Some(wm) = &self.wm_child
71 && let Ok(mut child) = wm.try_lock()
72 {
73 let _ = child.start_kill();
74 }
75 if let Ok(mut child) = self.child.try_lock() {
76 let _ = child.start_kill();
77 }
78 }
79}
80
81pub async fn launch_provider(
83 provider: TakeoverProviderKind,
84 quality: u8,
85) -> Result<TakeoverProxyState, Error> {
86 match provider {
87 TakeoverProviderKind::KasmVnc => Ok(TakeoverProxyState::new(
88 provider,
89 launch_kasmvnc_provider().await?,
90 quality,
91 )),
92 }
93}
94
95pub async fn launch_kasmvnc_provider() -> Result<Arc<ProviderHandle>, Error> {
98 #[cfg(not(unix))]
99 {
100 return Err(Error::new(
101 ErrorCode::BackendUnsupported,
102 "KasmVNC display provider is only supported on Unix-like container hosts",
103 ));
104 }
105
106 #[cfg(unix)]
107 {
108 launch_kasmvnc_unix().await
109 }
110}
111
112#[cfg(unix)]
113async fn launch_kasmvnc_unix() -> Result<Arc<ProviderHandle>, Error> {
114 let bin = resolve_kasmvnc_bin()?;
115 let web_root = resolve_kasmvnc_web_root()?;
116 let display_num = pick_display_number()?;
117 let display = format!(":{display_num}");
118 let web_port = pick_ephemeral_port().map_err(|e| {
119 Error::new(
120 ErrorCode::BrowserLaunchFailed,
121 format!("could not reserve KasmVNC web port: {e}"),
122 )
123 })?;
124 let rfb_port = pick_ephemeral_port().map_err(|e| {
125 Error::new(
126 ErrorCode::BrowserLaunchFailed,
127 format!("could not reserve KasmVNC VNC port: {e}"),
128 )
129 })?;
130
131 let mut cmd = Command::new(&bin);
132 cmd.arg(&display)
133 .arg("-geometry")
134 .arg(format!("{DISPLAY_WIDTH}x{DISPLAY_HEIGHT}"))
135 .arg("-depth")
136 .arg("24")
137 .arg("-interface")
138 .arg("127.0.0.1")
139 .arg("-rfbport")
140 .arg(rfb_port.to_string())
141 .arg("-websocketPort")
142 .arg(web_port.to_string())
143 .arg("-httpd")
144 .arg(web_root)
145 .arg("-sslOnly=0")
146 .arg("-SecurityTypes")
147 .arg("None")
148 .arg("-disableBasicAuth")
149 .arg("-PublicIP")
150 .arg("127.0.0.1")
151 .arg("-Log")
152 .arg("*:stderr:30")
153 .kill_on_drop(true)
154 .stdout(Stdio::null())
155 .stderr(Stdio::piped());
156
157 let mut child = cmd.spawn().map_err(|e| {
158 Error::new(
159 ErrorCode::BrowserLaunchFailed,
160 format!("spawn KasmVNC Xvnc {}: {e}", bin.display()),
161 )
162 })?;
163 let stderr_task = child.stderr.take().map(|stderr| {
164 tokio::spawn(async move {
165 let mut lines = tokio::io::BufReader::new(stderr).lines();
166 while matches!(lines.next_line().await, Ok(Some(_))) {}
167 })
168 });
169
170 if let Err(e) = wait_for_x_display(display_num, Duration::from_secs(10)).await {
171 let _ = child.start_kill();
172 return Err(Error::new(
173 ErrorCode::BrowserLaunchFailed,
174 format!("KasmVNC display {display} did not become ready: {e}"),
175 ));
176 }
177 if let Err(e) = wait_for_tcp_ready(("127.0.0.1", web_port), Duration::from_secs(10)).await {
178 let _ = child.start_kill();
179 return Err(Error::new(
180 ErrorCode::BrowserLaunchFailed,
181 format!("KasmVNC web client did not accept connections on port {web_port}: {e}"),
182 ));
183 }
184
185 let wm_child = spawn_window_manager(&display);
190
191 Ok(Arc::new(ProviderHandle {
192 display,
193 web_port,
194 window_manager: wm_child.is_some(),
195 _rfb_port: rfb_port,
196 child: Mutex::new(child),
197 wm_child: wm_child.map(Mutex::new),
198 stderr_task,
199 }))
200}
201
202#[cfg(unix)]
206fn spawn_window_manager(display: &str) -> Option<Child> {
207 for (bin, args) in [
208 ("matchbox-window-manager", &["-use_titlebar", "no"][..]),
209 ("openbox", &[][..]),
210 ] {
211 if resolve_named_bin(bin, &None).is_err() {
212 continue;
213 }
214 match Command::new(bin)
215 .args(args)
216 .env("DISPLAY", display)
217 .stdout(Stdio::null())
218 .stderr(Stdio::null())
219 .kill_on_drop(true)
220 .spawn()
221 {
222 Ok(child) => return Some(child),
223 Err(_) => continue,
224 }
225 }
226 None
227}
228
229fn resolve_kasmvnc_bin() -> Result<PathBuf, Error> {
230 if let Ok(raw) = std::env::var("AFHTTP_KASMVNC_BIN") {
231 let path = PathBuf::from(raw);
232 if path.exists() {
233 return Ok(path);
234 }
235 }
236 resolve_named_bin("Xvnc", &None).or_else(|_| resolve_named_bin("kasmvncserver", &None))
237}
238
239fn resolve_kasmvnc_web_root() -> Result<PathBuf, Error> {
240 if let Ok(raw) = std::env::var("AFHTTP_KASMVNC_WEB_ROOT") {
241 let path = PathBuf::from(raw);
242 if path.exists() {
243 return Ok(path);
244 }
245 }
246 for candidate in [
247 "/usr/share/kasmvnc/www",
248 "/usr/local/share/kasmvnc/www",
249 "/opt/kasmvnc/share/kasmvnc/www",
250 ] {
251 let path = PathBuf::from(candidate);
252 if path.exists() {
253 return Ok(path);
254 }
255 }
256 Err(Error::new(
257 ErrorCode::BrowserLaunchFailed,
258 "could not find KasmVNC web root; set AFHTTP_KASMVNC_WEB_ROOT",
259 ))
260}
261
262#[cfg(unix)]
263fn pick_display_number() -> Result<u16, Error> {
264 for display in 90..200 {
265 let socket = x_socket_path(display);
266 if !socket.exists() {
267 return Ok(display);
268 }
269 }
270 Err(Error::new(
271 ErrorCode::BrowserLaunchFailed,
272 "could not find a free X display number for KasmVNC",
273 ))
274}
275
276#[cfg(unix)]
277async fn wait_for_x_display(display: u16, timeout: Duration) -> Result<(), String> {
278 let deadline = tokio::time::Instant::now() + timeout;
279 let socket = x_socket_path(display);
280 loop {
281 if tokio::time::Instant::now() >= deadline {
282 return Err(format!("timed out after {timeout:?}"));
283 }
284 if UnixStream::connect(&socket).await.is_ok() {
285 return Ok(());
286 }
287 tokio::time::sleep(Duration::from_millis(50)).await;
288 }
289}
290
291#[cfg(unix)]
292fn x_socket_path(display: u16) -> PathBuf {
293 Path::new("/tmp/.X11-unix").join(format!("X{display}"))
294}
295
296#[derive(Clone)]
299pub struct TakeoverProxyState {
300 pub provider: TakeoverProviderKind,
301 pub display: String,
302 pub web_addr: SocketAddr,
303 pub window_manager: bool,
306 pub quality: u8,
308 _handle: Option<Arc<ProviderHandle>>,
309}
310
311impl TakeoverProxyState {
312 pub fn new(provider: TakeoverProviderKind, handle: Arc<ProviderHandle>, quality: u8) -> Self {
313 Self {
314 provider,
315 display: handle.display.clone(),
316 web_addr: SocketAddr::from(([127, 0, 0, 1], handle.web_port)),
317 window_manager: handle.window_manager,
318 quality,
319 _handle: Some(handle),
320 }
321 }
322
323 #[cfg(any(test, feature = "host"))]
324 pub fn for_tests(web_port: u16) -> Self {
325 Self {
326 provider: TakeoverProviderKind::KasmVnc,
327 display: ":99".to_string(),
328 web_addr: SocketAddr::from(([127, 0, 0, 1], web_port)),
329 window_manager: false,
330 quality: 100,
331 _handle: None,
332 }
333 }
334
335 pub async fn proxy(
337 self,
338 ws: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
339 uri: Uri,
340 request: axum::extract::Request,
341 ) -> Response {
342 match self.provider {
343 TakeoverProviderKind::KasmVnc => proxy_kasmvnc(self, ws, uri, request).await,
344 }
345 }
346}
347
348async fn proxy_kasmvnc(
349 display: TakeoverProxyState,
350 ws: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
351 uri: Uri,
352 request: axum::extract::Request,
353) -> Response {
354 let resize = if display.window_manager {
364 "remote"
365 } else {
366 "scale"
367 };
368 let is_landing = matches!(uri.path(), "/takeover/panel" | "/takeover/panel/");
369 let q = uri.query().unwrap_or("");
370 let has_path = q.split('&').any(|p| p.starts_with("path="));
371 let want_resize = format!("resize={resize}");
375 let resize_ok = q.split('&').any(|p| p == want_resize);
376 if is_landing && ws.is_err() && !(has_path && resize_ok) {
377 return landing_bootstrap(resize, display.quality, q);
378 }
379
380 let upstream_path = display_upstream_path_and_query(&uri);
381 if let Ok(ws) = ws {
382 let upstream = format!("ws://{}{}", display.web_addr, upstream_path);
383 return ws
388 .protocols(["binary"])
389 .on_upgrade(move |socket| forward_kasmvnc_ws(socket, upstream));
390 }
391
392 forward_display_http(display.web_addr, upstream_path, request).await
393}
394
395fn landing_bootstrap(resize: &str, quality: u8, query: &str) -> Response {
416 let mut settings = format!("&resize={resize}{}", kasmvnc_quality_params(quality));
421 for pair in query.split('&') {
422 if pair.is_empty()
423 || pair.starts_with("path=")
424 || pair.starts_with("resize=")
425 || pair.starts_with("handoff_secret=")
426 || pair.starts_with("handoff=")
427 {
428 continue;
429 }
430 settings.push('&');
431 settings.push_str(pair);
432 }
433 let body = format!(
434 "<!doctype html>\n<meta charset=\"utf-8\">\n<title>afhttp takeover</title>\n\
435 <noscript>The takeover panel needs JavaScript.</noscript>\n\
436 <script>\n(function () {{\n\
437 \x20 var path = location.pathname;\n\
438 \x20 var base = path.charAt(path.length - 1) === '/' ? path : path + '/';\n\
439 \x20 var websockify = base.replace(/^\\/+/, '') + 'websockify';\n\
440 \x20 location.replace(base + '?path=' + encodeURIComponent(websockify) + {settings});\n\
441 }})();\n</script>\n",
442 settings = js_string(&settings)
443 );
444 (
445 StatusCode::OK,
446 [
447 (header::CONTENT_TYPE, "text/html; charset=utf-8"),
448 (header::CACHE_CONTROL, "no-store"),
449 ],
450 body,
451 )
452 .into_response()
453}
454
455fn js_string(value: &str) -> String {
462 serde_json::to_string(value)
463 .unwrap_or_else(|_| "\"\"".to_string())
464 .replace('<', "\\u003c")
465}
466
467fn kasmvnc_quality_params(pct: u8) -> String {
476 let level = (u32::from(pct.min(100)) * 9 + 50) / 100; format!(
478 "&quality={level}&dynamic_quality_min={level}&dynamic_quality_max=9\
479 &jpeg_video_quality={level}&webp_video_quality={level}\
480 &max_video_resolution_x=3840&max_video_resolution_y=2160"
481 )
482}
483
484async fn forward_display_http(
485 upstream: std::net::SocketAddr,
486 path_and_query: String,
487 request: axum::extract::Request,
488) -> Response {
489 let (parts, body) = request.into_parts();
490 let url = format!("http://{upstream}{path_and_query}");
491 let method = reqwest::Method::from_bytes(parts.method.as_str().as_bytes())
492 .unwrap_or(reqwest::Method::GET);
493 let body = match to_bytes(body, 64 * 1024 * 1024).await {
494 Ok(bytes) => bytes,
495 Err(e) => {
496 return (
497 StatusCode::BAD_REQUEST,
498 format!("display proxy: could not read request body: {e}"),
499 )
500 .into_response();
501 }
502 };
503
504 let client = reqwest::Client::new();
505 let mut upstream_req = client.request(method, url).body(body);
506 for (name, value) in &parts.headers {
507 if should_forward_header(name) {
508 upstream_req = upstream_req.header(name, value);
509 }
510 }
511 upstream_req = upstream_req.header(header::HOST.as_str(), upstream.to_string());
512
513 let resp = match upstream_req.send().await {
514 Ok(resp) => resp,
515 Err(e) => {
516 return (
517 StatusCode::BAD_GATEWAY,
518 format!("display proxy: upstream request failed: {e}"),
519 )
520 .into_response();
521 }
522 };
523
524 let status = StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
525 let mut headers = HeaderMap::new();
526 for (name, value) in resp.headers() {
527 if should_forward_response_header(name) {
528 headers.insert(name.clone(), value.clone());
529 }
530 }
531 let bytes = match resp.bytes().await {
532 Ok(bytes) => bytes,
533 Err(e) => {
534 return (
535 StatusCode::BAD_GATEWAY,
536 format!("display proxy: upstream body failed: {e}"),
537 )
538 .into_response();
539 }
540 };
541
542 (status, headers, Body::from(bytes)).into_response()
543}
544
545async fn forward_kasmvnc_ws(client: WebSocket, upstream_url: String) {
546 use tungstenite::client::IntoClientRequest;
547 let (mut client_tx, mut client_rx) = client.split();
548 let upstream_req = match upstream_url.as_str().into_client_request() {
551 Ok(mut req) => {
552 req.headers_mut().insert(
553 tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL,
554 tungstenite::http::HeaderValue::from_static("binary"),
555 );
556 if let Some(authority) = req.uri().authority().map(|a| a.to_string())
561 && let Ok(origin) =
562 tungstenite::http::HeaderValue::from_str(&format!("http://{authority}"))
563 {
564 req.headers_mut()
565 .insert(tungstenite::http::header::ORIGIN, origin);
566 }
567 req
568 }
569 Err(_) => {
570 let _ = client_tx.close().await;
571 return;
572 }
573 };
574 let upstream_stream = match tokio_tungstenite::connect_async(upstream_req).await {
575 Ok((stream, _resp)) => stream,
576 Err(_) => {
577 let _ = client_tx.close().await;
578 return;
579 }
580 };
581 let (mut upstream_tx, mut upstream_rx) = upstream_stream.split();
582
583 let c2u = async {
584 while let Some(Ok(msg)) = client_rx.next().await {
585 let outbound = match msg {
586 Message::Text(t) => tungstenite::Message::Text(t.as_str().into()),
587 Message::Binary(b) => tungstenite::Message::Binary(b.to_vec().into()),
588 Message::Ping(p) => tungstenite::Message::Ping(p.to_vec().into()),
589 Message::Pong(p) => tungstenite::Message::Pong(p.to_vec().into()),
590 Message::Close(_) => break,
591 };
592 if upstream_tx.send(outbound).await.is_err() {
593 break;
594 }
595 }
596 let _ = upstream_tx.send(tungstenite::Message::Close(None)).await;
597 };
598 let u2c = async {
599 while let Some(Ok(msg)) = upstream_rx.next().await {
600 let inbound = match msg {
601 tungstenite::Message::Text(t) => Message::Text(t.as_str().into()),
602 tungstenite::Message::Binary(b) => Message::Binary(b.to_vec().into()),
603 tungstenite::Message::Ping(p) => Message::Ping(p.to_vec().into()),
604 tungstenite::Message::Pong(p) => Message::Pong(p.to_vec().into()),
605 tungstenite::Message::Close(_) => break,
606 tungstenite::Message::Frame(_) => continue,
607 };
608 if client_tx.send(inbound).await.is_err() {
609 break;
610 }
611 }
612 let _ = client_tx.close().await;
613 };
614 tokio::pin!(c2u);
615 tokio::pin!(u2c);
616 tokio::select! {
617 _ = &mut c2u => {},
618 _ = &mut u2c => {},
619 }
620}
621
622fn display_upstream_path_and_query(uri: &Uri) -> String {
623 let suffix = uri
624 .path()
625 .strip_prefix("/takeover/panel")
626 .filter(|s| !s.is_empty())
627 .unwrap_or("/");
628 let path = if suffix.starts_with('/') {
629 suffix.to_string()
630 } else {
631 format!("/{suffix}")
632 };
633 let Some(query) = uri.query() else {
634 return path;
635 };
636 let filtered: Vec<&str> = query
637 .split('&')
638 .filter(|pair| {
639 !pair.starts_with("handoff_secret=")
640 && !pair.starts_with("handoff=")
641 && !pair.starts_with("token_secret=")
642 })
643 .collect();
644 if filtered.is_empty() {
645 path
646 } else {
647 format!("{path}?{}", filtered.join("&"))
648 }
649}
650
651fn should_forward_header(name: &header::HeaderName) -> bool {
652 !matches!(
653 name.as_str().to_ascii_lowercase().as_str(),
654 "host"
655 | "connection"
656 | "upgrade"
657 | "content-length"
658 | "sec-websocket-key"
659 | "sec-websocket-version"
660 | "sec-websocket-protocol"
661 | "sec-websocket-extensions"
662 | "authorization"
663 | "cookie"
664 )
665}
666
667fn should_forward_response_header(name: &header::HeaderName) -> bool {
668 !matches!(
669 name.as_str().to_ascii_lowercase().as_str(),
670 "connection" | "transfer-encoding" | "upgrade" | "content-length"
671 )
672}
673
674#[cfg(test)]
675mod tests {
676 use super::{js_string, kasmvnc_quality_params};
677
678 #[test]
679 fn a_seeded_setting_cannot_close_the_script_it_is_written_in() {
680 let escaped = js_string("&x=</script><img src=x onerror=alert(1)>");
683 assert!(!escaped.contains("</script>"), "{escaped}");
684 assert!(
685 escaped.starts_with('"') && escaped.ends_with('"'),
686 "{escaped}"
687 );
688 assert_eq!(js_string("&resize=scale"), "\"&resize=scale\"");
689 assert_eq!(js_string("a\"b\\c"), "\"a\\\"b\\\\c\"");
690 }
691
692 #[test]
693 fn quality_pct_maps_to_kasmvnc_tiers() {
694 assert!(kasmvnc_quality_params(100).contains("&quality=9&"));
695 assert!(kasmvnc_quality_params(0).contains("&quality=0&"));
696 assert!(kasmvnc_quality_params(50).contains("&quality=5&"));
697 assert!(kasmvnc_quality_params(200).contains("&quality=9&"));
699 assert!(kasmvnc_quality_params(10).contains("max_video_resolution_x=3840"));
701 }
702}