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