@@ -1,28 +1,23 @@
11use std::{
22 env, fs,
3− io::{self, Write},
3+ io::{self},
44 net::{SocketAddr, TcpStream},
5− os::unix::{fs::PermissionsExt, io::OwnedFd, net::UnixStream},
5+ os::unix::{fs::PermissionsExt, net::UnixStream},
66 path::{Path, PathBuf},
77 process::{Command, Stdio},
8− sync::{
9− Arc, Mutex,
10− atomic::{AtomicBool, Ordering},
11− },
8+ sync::{Arc, Mutex},
129 thread,
1310 time::{Duration, Instant},
1411};
1512
1613use serde::{Deserialize, Serialize};
17−use std::sync::mpsc::Sender;
1814use time::{OffsetDateTime, format_description::well_known::Rfc3339};
1915use uuid::Uuid;
2016
2117use crate::{
2218 commands,
2319 session_manager::{INSTANCE_DIR_NAME, INSTANCE_FILENAME},
24− tui::AppState,
25− vm::{self, LoginAction, VmInput},
20+ vm::{self, LoginAction},
2621};
2722
2823const SSH_KEY_NAME: &str = "ssh_key";
@@ -422,243 +417,3 @@ pub(crate) fn build_ssh_login_actions(
422417
423418 vec![LoginAction::Send(setup)]
424419}
425−
426−fn spawn_ssh_io(
427− app: Arc<Mutex<AppState>>,
428− config: Arc<Mutex<InstanceConfig>>,
429− instance_dir: PathBuf,
430− ssh_key: PathBuf,
431− output_monitor: Arc<vm::OutputMonitor>,
432− vm_output_fd: OwnedFd,
433− vm_input_fd: OwnedFd,
434−) -> vm::IoContext {
435− let io_control = vm::IoControl::new();
436−
437− let log_path = instance_dir.join(VM_ROOT_LOG_NAME);
438− let log_file = fs::OpenOptions::new()
439− .create(true)
440− .write(true)
441− .truncate(true)
442− .open(&log_path)
443− .ok()
444− .map(|file| Arc::new(Mutex::new(file)));
445−
446− let ssh_connected = Arc::new(AtomicBool::new(false));
447− let ssh_started = Arc::new(AtomicBool::new(false));
448− let ssh_ready = Arc::new(AtomicBool::new(false));
449− let input_tx_holder: Arc<Mutex<Option<Sender<VmInput>>>> = Arc::new(Mutex::new(None));
450−
451− let instance_path = instance_dir.join(INSTANCE_FILENAME);
452− let config_for_output = config.clone();
453− let log_for_output = log_file.clone();
454− let ssh_connected_for_output = ssh_connected.clone();
455− let ssh_ready_for_output = ssh_ready.clone();
456−
457− let mut line_buf = String::new();
458−
459− let handlers = commands::build_handlers(app.clone(), io_control.clone());
460− let on_line = move |line: &str| handlers.handle(line);
461−
462− let on_output = move |bytes: &[u8]| {
463− if ssh_connected_for_output.load(Ordering::SeqCst) {
464− if let Some(log) = &log_for_output {
465− if let Ok(mut file) = log.lock() {
466− let _ = file.write_all(bytes);
467− }
468− }
469− }
470−
471− let text = String::from_utf8_lossy(bytes);
472− line_buf.push_str(&text);
473−
474− while let Some(pos) = line_buf.find('\n') {
475− let mut line = line_buf[..pos].to_string();
476− if line.ends_with('\r') {
477− line.pop();
478− }
479− line_buf.drain(..=pos);
480−
481− let cleaned = line.trim_start_matches(|c: char| c == '\r' || c == ' ');
482−
483− if let Some(pos) = cleaned.find("VIBEBOX_IPV4=") {
484− let ip_raw = &cleaned[(pos + "VIBEBOX_IPV4=".len())..];
485− let ip = extract_ipv4(ip_raw).unwrap_or_default();
486− if !ip.is_empty() {
487− if let Ok(mut cfg) = config_for_output.lock() {
488− if cfg.vm_ipv4.as_deref() != Some(ip.as_str()) {
489− cfg.vm_ipv4 = Some(ip.clone());
490− let _ = write_instance_config(&instance_path, &cfg);
491− tracing::info!(ip = %ip, "detected vm ipv4");
492− }
493− }
494− }
495− }
496−
497− if cleaned.contains("VIBEBOX_SSH_READY") {
498− ssh_ready_for_output.store(true, Ordering::SeqCst);
499− tracing::info!("sshd ready");
500− }
501− }
502− };
503−
504− let io_ctx = vm::spawn_vm_io_with_hooks(
505− output_monitor,
506− vm_output_fd,
507− vm_input_fd,
508− io_control.clone(),
509− on_line,
510− on_output,
511− );
512−
513− *input_tx_holder.lock().unwrap() = Some(io_ctx.input_tx.clone());
514−
515− let ssh_ready_for_thread = ssh_ready.clone();
516− let ssh_started_for_thread = ssh_started.clone();
517− let ssh_connected_for_thread = ssh_connected.clone();
518− let io_control_for_thread = io_control.clone();
519− let ssh_key_for_thread = ssh_key.clone();
520− let config_for_thread = config.clone();
521− let input_tx_holder_for_thread = input_tx_holder.clone();
522−
523− thread::spawn(move || {
524− loop {
525− if ssh_started_for_thread.load(Ordering::SeqCst) {
526− break;
527− }
528− if !ssh_ready_for_thread.load(Ordering::SeqCst) {
529− thread::sleep(std::time::Duration::from_millis(200));
530− continue;
531− }
532−
533− let ip = config_for_thread
534− .lock()
535− .ok()
536− .and_then(|cfg| cfg.vm_ipv4.clone());
537− let ssh_user = config_for_thread
538− .lock()
539− .map(|cfg| cfg.ssh_user.clone())
540− .unwrap_or_else(|_| DEFAULT_SSH_USER.to_string());
541−
542− if let Some(ip) = ip {
543− if ssh_started_for_thread
544− .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
545− .is_err()
546− {
547− break;
548− }
549−
550− tracing::info!(user = %ssh_user, ip = %ip, "starting ssh");
551− io_control_for_thread.request_terminal_restore();
552− io_control_for_thread.set_forward_output(false);
553− io_control_for_thread.set_forward_input(false);
554− ssh_connected_for_thread.store(true, Ordering::SeqCst);
555−
556− let mut attempts = 0usize;
557− loop {
558− attempts += 1;
559− if !ssh_port_open(&ip) {
560− tracing::info!(
561− attempts,
562− ip = %ip,
563− "waiting for ssh port ({}/{})",
564− attempts,
565− SSH_CONNECT_RETRIES
566− );
567− if attempts >= SSH_CONNECT_RETRIES {
568− ssh_connected_for_thread.store(false, Ordering::SeqCst);
569− tracing::warn!(
570− attempts,
571− "ssh port not ready after {} attempts",
572− SSH_CONNECT_RETRIES
573− );
574− break;
575− }
576− thread::sleep(std::time::Duration::from_millis(SSH_CONNECT_DELAY_MS));
577− continue;
578− }
579−
580− tracing::info!(
581− attempts,
582− user = %ssh_user,
583− ip = %ip,
584− "starting ssh ({}/{})",
585− attempts,
586− SSH_CONNECT_RETRIES
587− );
588− let status = Command::new("ssh")
589− .args([
590− "-i",
591− ssh_key_for_thread.to_str().unwrap_or(".vibebox/ssh_key"),
592− "-o",
593− "IdentitiesOnly=yes",
594− "-o",
595− "StrictHostKeyChecking=no",
596− "-o",
597− "UserKnownHostsFile=/dev/null",
598− "-o",
599− "GlobalKnownHostsFile=/dev/null",
600− "-o",
601− "PasswordAuthentication=no",
602− "-o",
603− "BatchMode=yes",
604− "-o",
605− "LogLevel=ERROR",
606− "-o",
607− "ConnectTimeout=5",
608− ])
609− .env_remove("LC_CTYPE")
610− .env_remove("LC_ALL")
611− // .env_remove("LANG")
612− .arg(format!("{ssh_user}@{ip}"))
613− .stdin(Stdio::inherit())
614− .stdout(Stdio::inherit())
615− .stderr(Stdio::inherit())
616− .status();
617−
618− match status {
619− Ok(status) if status.success() => {
620− ssh_connected_for_thread.store(false, Ordering::SeqCst);
621− tracing::info!(status = %status, "ssh exited");
622− if let Some(tx) = input_tx_holder_for_thread.lock().unwrap().clone() {
623− let _ = tx.send(VmInput::Bytes(b"systemctl poweroff\n".to_vec()));
624− }
625− break;
626− }
627− Ok(status) if status.code() == Some(255) => {
628− tracing::warn!(status = %status, "ssh connection failed");
629− if attempts >= SSH_CONNECT_RETRIES {
630− ssh_connected_for_thread.store(false, Ordering::SeqCst);
631− tracing::warn!(
632− attempts,
633− "ssh failed after {} attempts",
634− SSH_CONNECT_RETRIES
635− );
636− break;
637− }
638− thread::sleep(std::time::Duration::from_millis(500));
639− continue;
640− }
641− Ok(status) => {
642− ssh_connected_for_thread.store(false, Ordering::SeqCst);
643− tracing::info!(status = %status, "ssh exited");
644− if let Some(tx) = input_tx_holder_for_thread.lock().unwrap().clone() {
645− let _ = tx.send(VmInput::Bytes(b"systemctl poweroff\n".to_vec()));
646− }
647− break;
648− }
649− Err(err) => {
650− ssh_connected_for_thread.store(false, Ordering::SeqCst);
651− tracing::error!(error = %err, "failed to start ssh");
652− break;
653− }
654− }
655− }
656− break;
657− }
658−
659− thread::sleep(std::time::Duration::from_millis(200));
660− }
661− });
662−
663− io_ctx
664−}