+11

Cargo.lock

View file Reviewed

··· 6217 6217 ] 6218 6218 6219 6219 [[package]] 6220 6220 + name = "pty-process" 6221 6221 + version = "0.5.3" 6222 6222 + source = "registry+https://github.com/rust-lang/crates.io-index" 6223 6223 + checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266" 6224 6224 + dependencies = [ 6225 6225 + "rustix", 6226 6226 + "tokio", 6227 6227 + ] 6228 6228 + 6229 6229 + [[package]] 6220 6230 name = "quanta" 6221 6231 version = "0.12.6" 6222 6232 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 7417 7427 "prost", 7418 7428 "prost-protovalidate", 7419 7429 "prost-reflect", 7430 7430 + "pty-process", 7420 7431 "serde", 7421 7432 "serde_json", 7422 7433 "tempfile",

+15

docker-compose.mill.yml

View file Reviewed

··· 31 31 SPINDLE_NIX_CACHE_READ_URLS: http://ncps:8501 32 32 SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS: cache.local:F7YqpMzuBdILYd/v+wMZN2YKxCzliXQyFmeezOxw7rU= 33 33 SPINDLE_NIX_CACHE_UPLOAD_URL: http://ncps:8501/upload 34 34 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED: "true" 35 35 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2223 36 36 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: 127.0.0.1 37 37 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_JUMP_HOST: chernobog 38 38 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD: 10m 39 39 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST_KEY_PATH: /var/lib/spindle/debug_ssh_host_key 34 40 # dials the mill container directly using ws 35 41 SPINDLE_MILL_URL: ws://spindle:6555/mill 36 42 devices: ··· 113 119 SPINDLE_MILL_LABELS: linux,fast 114 120 SPINDLE_MILL_TOKEN_FILE: /shared/executor-a.mill-token 115 121 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11241" 122 122 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2224 123 123 + ports: 124 124 + - "127.0.0.1:2224:2224" 116 125 volumes: 117 126 - spindle-executor-a-data:/var/lib/spindle 118 127 - spindle-artifacts:/var/lib/spindle/artifacts ··· 128 137 SPINDLE_MILL_LABELS: linux,slow 129 138 SPINDLE_MILL_TOKEN_FILE: /shared/executor-b.mill-token 130 139 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11242" 140 140 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2225 141 141 + ports: 142 142 + - "127.0.0.1:2225:2225" 131 143 volumes: 132 144 - spindle-executor-b-data:/var/lib/spindle 133 145 - spindle-artifacts:/var/lib/spindle/artifacts ··· 143 155 SPINDLE_MILL_LABELS: linux,gpu 144 156 SPINDLE_MILL_TOKEN_FILE: /shared/executor-c.mill-token 145 157 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11243" 158 158 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2226 159 159 + ports: 160 160 + - "127.0.0.1:2226:2226" 146 161 volumes: 147 162 - spindle-executor-c-data:/var/lib/spindle 148 163 - spindle-artifacts:/var/lib/spindle/artifacts

+5 -3

docker-compose.yml

View file Reviewed

··· 182 182 SPINDLE_MICROVM_PIPELINES_OVERLAY_DIR: /var/lib/spindle/overlays 183 183 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11240" 184 184 SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS: "false" 185 185 - # route guest nix substitution + uploads through the local ncps cache. 186 186 - # ncps re-signs on serve with cache.local's key, so the guest trusts the 187 187 - # matching public key below (no signing happens in spindle itself). 188 185 SPINDLE_NIX_CACHE_READ_URLS: http://ncps:8501 189 186 SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS: cache.local:F7YqpMzuBdILYd/v+wMZN2YKxCzliXQyFmeezOxw7rU= 190 187 SPINDLE_NIX_CACHE_UPLOAD_URL: http://ncps:8501/upload 188 188 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED: true 189 189 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2223 190 190 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD: 10m 191 191 # these two are required for cgroups, uncomment if testing 192 192 # privileged: true 193 193 # cgroup: host ··· 202 202 security_opt: 203 203 - label=disable 204 204 - seccomp=unconfined 205 205 + ports: 206 206 + - "2223:2223" 205 207 volumes: 206 208 - spindle-data:/var/lib/spindle 207 209 - spindle-logs:/var/log/spindle

+2 -1

shuttle/Cargo.toml

View file Reviewed

··· 8 8 [dependencies] 9 9 anyhow = "1" 10 10 base64 = "0.22" 11 11 - nix = { version = "0.31", features = ["fs", "process", "reboot", "signal", "user"] } 11 11 + nix = { version = "0.31", features = ["fs", "process", "reboot", "signal", "term", "user"] } 12 12 prost = "0.14" 13 13 prost-reflect = "0.16" 14 14 prost-protovalidate = "0.3" 15 15 once_cell = "1" 16 16 + pty-process = { version = "0.5.3", features = ["async"] } 16 17 serde = { version = "1", features = ["derive"] } 17 18 serde_json = "1" 18 19 tempfile = "3"

+1

nix/microvm/base.nix

View file Reviewed

··· 199 199 group = "spindle-workflow"; 200 200 home = "/workspace"; 201 201 createHome = false; 202 202 + shell = pkgs.bashInteractive; 202 203 }; 203 204 users.users.spindle-workflow.extraGroups = lib.mkIf config.virtualisation.docker.enable [ 204 205 "docker"

+51

nix/modules/spindle.nix

View file Reviewed

··· 278 278 ''; 279 279 }; 280 280 }; 281 281 + 282 282 + debugSsh = { 283 283 + enable = mkOption { 284 284 + type = types.bool; 285 285 + default = false; 286 286 + description = '' 287 287 + Enable the debug ssh server that lets authorized users ssh into a 288 288 + failed microVM to debug it. 289 289 + ''; 290 290 + }; 291 291 + listenAddr = mkOption { 292 292 + type = types.str; 293 293 + default = "0.0.0.0:2222"; 294 294 + example = "0.0.0.0:2225"; 295 295 + description = "Address for the debug ssh server to listen on."; 296 296 + }; 297 297 + host = mkOption { 298 298 + type = types.str; 299 299 + default = ""; 300 300 + example = "127.0.0.1"; 301 301 + description = "Host reached from the SSH jump host."; 302 302 + }; 303 303 + jumpHost = mkOption { 304 304 + type = types.str; 305 305 + default = ""; 306 306 + example = "spindle.example.com"; 307 307 + description = "SSH jump host used in the printed debug command."; 308 308 + }; 309 309 + hostKeyPath = mkOption { 310 310 + type = with types; nullOr path; 311 311 + default = null; 312 312 + example = "/var/lib/spindle/debug_ssh_host_key"; 313 313 + description = '' 314 314 + Path to the ssh host key for the debug server. If null, one is generated 315 315 + once and persisted next to the spindle db. 316 316 + ''; 317 317 + }; 318 318 + gracePeriod = mkOption { 319 319 + type = types.str; 320 320 + default = "5m"; 321 321 + description = '' 322 322 + How long a failed workflow's microVM is kept alive for the user to ssh in. 323 323 + ''; 324 324 + }; 325 325 + }; 281 326 }; 282 327 283 328 nixCache = { ··· 400 445 "SPINDLE_MICROVM_PIPELINES_CGROUP_PIDS_MAX=${toString cfg.pipelines.microvm.cgroup.pidsMax}" 401 446 "SPINDLE_MICROVM_PIPELINES_CGROUP_SWAP_MAX_MIB=${toString cfg.pipelines.microvm.cgroup.swapMaxMiB}" 402 447 "SPINDLE_MICROVM_PIPELINES_CGROUP_SUPERVISOR_MEMORY_MIN_MIB=${toString cfg.pipelines.microvm.cgroup.supervisorMinMiB}" 448 448 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED=${lib.boolToString cfg.pipelines.microvm.debugSsh.enable}" 449 449 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR=${cfg.pipelines.microvm.debugSsh.listenAddr}" 450 450 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST=${cfg.pipelines.microvm.debugSsh.host}" 451 451 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_JUMP_HOST=${cfg.pipelines.microvm.debugSsh.jumpHost}" 452 452 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST_KEY_PATH=${optionalString (cfg.pipelines.microvm.debugSsh.hostKeyPath != null) (toString cfg.pipelines.microvm.debugSsh.hostKeyPath)}" 453 453 + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD=${cfg.pipelines.microvm.debugSsh.gracePeriod}" 403 454 "SPINDLE_NIX_CACHE_READ_URLS=${concatStringsSep "," cfg.pipelines.nixCache.readUrls}" 404 455 "SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS=${concatStringsSep "," cfg.pipelines.nixCache.trustedPublicKeys}" 405 456 "SPINDLE_NIX_CACHE_UPLOAD_URL=${cfg.pipelines.nixCache.uploadUrl}"

+99 -14

shuttle/src/command.rs

View file Reviewed

··· 1 1 use anyhow::{Context, Result}; 2 2 use nix::sys::signal::{Signal, kill}; 3 3 use nix::unistd::{Gid, Pid, Uid, User, getgrouplist, setgid, setgroups, setuid}; 4 4 - use std::ffi::{CString, OsStr, OsString}; 4 4 + use pty_process::{Command as PtyCommand, OwnedReadPty, OwnedWritePty, Size}; 5 5 + use std::ffi::{CString, OsString}; 5 6 use std::io; 6 7 use std::os::unix::process::ExitStatusExt; 7 8 use std::path::PathBuf; ··· 215 216 // group won't actually let it access the sock. 216 217 // https://github.com/rust-lang/rust/issues/90747 217 218 if let (Some(uid), Some(gid)) = (spec.uid, spec.gid) { 218 218 - let username = User::from_uid(Uid::from_raw(uid)) 219 219 - .ok() 220 220 - .flatten() 221 221 - .map(|u| u.name) 222 222 - .with_context(|| format!("lookup passwd entry for uid {uid}"))?; 223 223 - let cname = CString::new(username) 224 224 - .with_context(|| format!("username for uid {uid} contained a null byte"))?; 225 225 - // resolve groups beforehand so we don't have to read /etc/group in the pre_exec 226 226 - let groups = 227 227 - getgrouplist(&cname, Gid::from_raw(gid)).context("resolve supplementary groups")?; 219 219 + let groups = resolve_supplementary_groups(uid, gid)?; 228 220 // SAFETY: pre_exec runs between fork and execve in the child. 229 221 // we only call async-signal-safe syscalls and we don't touch any 230 222 // shared state, no allocator, no mutexes, no globals. ··· 242 234 cmd.process_group(0); 243 235 244 236 cmd.spawn() 245 245 - .with_context(|| format!("spawn {}", display_os(&spec.program))) 237 237 + .with_context(|| format!("spawn {:?}", &spec.program)) 238 238 + } 239 239 + 240 240 + // resolve the supplementary group list up front so the pre_exec hook never has 241 241 + // to read /etc/group (which is not async-signal-safe) between fork and exec. 242 242 + fn resolve_supplementary_groups(uid: u32, gid: u32) -> Result<Vec<Gid>> { 243 243 + let username = User::from_uid(Uid::from_raw(uid)) 244 244 + .ok() 245 245 + .flatten() 246 246 + .map(|u| u.name) 247 247 + .with_context(|| format!("lookup passwd entry for uid {uid}"))?; 248 248 + let cname = CString::new(username) 249 249 + .with_context(|| format!("username for uid {uid} contained a null byte"))?; 250 250 + getgrouplist(&cname, Gid::from_raw(gid)).context("resolve supplementary groups") 251 251 + } 252 252 + 253 253 + pub fn spawn_pty(spec: Spec, rows: u16, cols: u16) -> Result<(OwnedReadPty, OwnedWritePty, Child)> { 254 254 + let (pty, pts) = pty_process::open().context("open pty")?; 255 255 + pty.resize(Size::new(rows, cols)).context("set pty size")?; 256 256 + 257 257 + let mut cmd = PtyCommand::new(&spec.program) 258 258 + .args(&spec.args) 259 259 + .envs(spec.env.iter().map(|(key, value)| (key, value))); 260 260 + if let Some(cwd) = &spec.cwd { 261 261 + cmd = cmd.current_dir(cwd); 262 262 + } 263 263 + 264 264 + // drop privileges in the child. this RELIES on pty-process composing our 265 265 + // pre_exec hook *after* its own session setup: it wraps us as `move || { 266 266 + // session_leader()?; ours()?; }`, so setsid + TIOCSCTTY run first (while 267 267 + // still privileged) and only then do we drop to the workflow user. that 268 268 + // ordering is what we want and we depend on it. if pty-process ever ran our 269 269 + // hook first, the session setup would happen post-drop. (it'd likely still 270 270 + // work, since setsid/TIOCSCTTY on our own pty need no privilege, but it is 271 271 + // not the behaviour we're assuming here) 272 272 + // don't use .uid()/.gid() here, they clear supplementary groups (see L195). 273 273 + if let (Some(uid), Some(gid)) = (spec.uid, spec.gid) { 274 274 + let groups = resolve_supplementary_groups(uid, gid)?; 275 275 + // SAFETY: pre_exec runs between fork and execve in the child. every call 276 276 + // below is async-signal-safe and touches no shared state. 277 277 + cmd = unsafe { 278 278 + cmd.pre_exec(move || { 279 279 + setgroups(&groups).map_err(io::Error::from)?; 280 280 + setgid(Gid::from_raw(gid)).map_err(io::Error::from)?; 281 281 + setuid(Uid::from_raw(uid)).map_err(io::Error::from)?; 282 282 + Ok(()) 283 283 + }) 284 284 + }; 285 285 + } 286 286 + 287 287 + // spawn consumes the slave (dup'd onto the child's 0/1/2 and then closed in 288 288 + // the parent), so the master reports EOF once the shell and all its children 289 289 + // have exited. 290 290 + let child = cmd 291 291 + .spawn(pts) 292 292 + .with_context(|| format!("spawn pty shell {:?}", &spec.program))?; 293 293 + 294 294 + let (reader, writer) = pty.into_split(); 295 295 + Ok((reader, writer, child)) 246 296 } 247 297 248 298 async fn wait_child( ··· 327 377 }) 328 378 } 329 379 330 330 - fn display_os(value: &OsStr) -> String { 331 331 - value.to_string_lossy().into_owned() 380 380 + #[cfg(test)] 381 381 + mod tests { 382 382 + use super::*; 383 383 + use tokio::io::AsyncReadExt; 384 384 + 385 385 + #[tokio::test] 386 386 + async fn pty_runs_a_shell_and_reports_exit() { 387 387 + let spec = Spec::new("/bin/sh") 388 388 + .arg("-c") 389 389 + .arg("printf 'hello pty'; exit 7"); 390 390 + let (mut reader, _writer, mut child) = spawn_pty(spec, 24, 80).expect("spawn pty"); 391 391 + 392 392 + let mut output = Vec::new(); 393 393 + let mut chunk = [0u8; 1024]; 394 394 + loop { 395 395 + match reader.read(&mut chunk).await { 396 396 + Ok(0) => break, 397 397 + Ok(n) => output.extend_from_slice(&chunk[..n]), 398 398 + // linux signals slave-closed with EIO rather than EOF 399 399 + Err(error) if error.raw_os_error() == Some(nix::libc::EIO) => break, 400 400 + Err(error) => panic!("read pty master: {error}"), 401 401 + } 402 402 + } 403 403 + 404 404 + let status = child.wait().await.expect("wait child"); 405 405 + let text = String::from_utf8_lossy(&output); 406 406 + assert!(text.contains("hello pty"), "unexpected output: {text:?}"); 407 407 + assert_eq!(status.code(), Some(7)); 408 408 + } 409 409 + 410 410 + #[tokio::test] 411 411 + async fn pty_resize_succeeds() { 412 412 + let spec = Spec::new("/bin/sh").arg("-c").arg("sleep 0.2"); 413 413 + let (_reader, writer, mut child) = spawn_pty(spec, 24, 80).expect("spawn pty"); 414 414 + writer.resize(Size::new(40, 120)).expect("resize"); 415 415 + let _ = child.wait().await; 416 416 + } 332 417 }

+1

shuttle/src/main.rs

View file Reviewed

··· 9 9 mod logging; 10 10 mod nix_config; 11 11 mod protocol; 12 12 + mod pty; 12 13 mod session; 13 14 14 15 use std::env;

+6

shuttle/src/protocol.rs

View file Reviewed

··· 43 43 CacheDrainResult, 44 44 Poweroff, 45 45 PoweroffResult, 46 46 + OpenDebugShell, 47 47 + PtyData, 48 48 + PtyResize, 46 49 Message, 47 50 ); 48 51 ··· 79 82 cache_drain_result => "cache_drain_result", 80 83 poweroff => "poweroff", 81 84 poweroff_result => "poweroff_result", 85 85 + open_debug_shell => "open_debug_shell", 86 86 + pty_data => "pty_data", 87 87 + pty_resize => "pty_resize", 82 88 }) 83 89 .unwrap_or_else(|| unreachable!("validated message has no payload")) 84 90 }

+182

shuttle/src/pty.rs

View file Reviewed

··· 1 1 + use std::path::Path; 2 2 + 3 3 + use crate::command::{self, Spec}; 4 4 + use crate::protocol::{self, Message, v1}; 5 5 + use anyhow::{Context, Result, bail}; 6 6 + use nix::sys::signal::{Signal, kill}; 7 7 + use nix::unistd::{Pid, User}; 8 8 + use pty_process::Size; 9 9 + use tokio::io::{AsyncReadExt, BufReader}; 10 10 + use tokio_vsock::{VsockAddr, VsockStream}; 11 11 + use tracing::{info, warn}; 12 12 + 13 13 + const WF_USER: &str = "spindle-workflow"; 14 14 + const READ_CHUNK: usize = 32 * 1024; 15 15 + 16 16 + pub async fn run(host_cid: u32, open: v1::OpenDebugShell) { 17 17 + if let Err(error) = serve(host_cid, open).await { 18 18 + warn!(%error, "debug shell session failed"); 19 19 + } 20 20 + } 21 21 + 22 22 + async fn serve(host_cid: u32, open: v1::OpenDebugShell) -> Result<()> { 23 23 + let conn = VsockStream::connect(VsockAddr::new(host_cid, open.vsock_port)) 24 24 + .await 25 25 + .with_context(|| format!("dial host debug vsock port {}", open.vsock_port))?; 26 26 + info!(port = open.vsock_port, "debug shell connected"); 27 27 + 28 28 + let user = resolve_user(WF_USER)?; 29 29 + 30 30 + let rows = clamp_tty_dim(open.rows); 31 31 + let cols = clamp_tty_dim(open.cols); 32 32 + 33 33 + let spec = Spec::new(&user.shell) 34 34 + .arg("-l") 35 35 + .envs(user.env(&open.term)) 36 36 + .run_as(user.uid, user.gid) 37 37 + .cwd(Path::new(&user.home).join("repo")); // /workflow/repo 38 38 + 39 39 + let (mut pty_reader, mut pty_writer, mut child) = 40 40 + command::spawn_pty(spec, rows, cols).context("spawn pty shell")?; 41 41 + let pid = child.id(); 42 42 + 43 43 + let (conn_reader, conn_writer) = tokio::io::split(conn); 44 44 + let mut conn_reader = BufReader::new(conn_reader); 45 45 + let mut conn_writer = conn_writer; 46 46 + 47 47 + let mut buf = vec![0u8; READ_CHUNK]; 48 48 + let client_gone = loop { 49 49 + tokio::select! { 50 50 + read = pty_reader.read(&mut buf) => match read { 51 51 + Ok(0) => break false, // shell exited 52 52 + Ok(n) => { 53 53 + let msg = Message { 54 54 + id: "pty".to_owned(), 55 55 + pty_data: Some(v1::PtyData { data: buf[..n].to_vec().into() }), 56 56 + ..Default::default() 57 57 + }; 58 58 + if protocol::write_message(&mut conn_writer, &msg).await.is_err() { 59 59 + break true; 60 60 + } 61 61 + } 62 62 + Err(error) => { 63 63 + // linux returns EIO (not a clean EOF) on the master once the 64 64 + // slave side is fully closed, so treat that as the shell 65 65 + // exiting normally rather than a real read failure. 66 66 + if error.raw_os_error() != Some(nix::libc::EIO) { 67 67 + warn!(%error, "pty master read failed"); 68 68 + } 69 69 + break false; 70 70 + } 71 71 + }, 72 72 + incoming = protocol::read_message(&mut conn_reader) => match incoming { 73 73 + Ok(Some(msg)) => { 74 74 + if let Some(data) = msg.pty_data { 75 75 + use tokio::io::AsyncWriteExt; 76 76 + if pty_writer.write_all(&data.data).await.is_err() { 77 77 + break false; 78 78 + } 79 79 + } else if let Some(resize) = msg.pty_resize { 80 80 + let size = Size::new(clamp_tty_dim(resize.rows), clamp_tty_dim(resize.cols)); 81 81 + if let Err(error) = pty_writer.resize(size) { 82 82 + warn!(%error, "pty resize failed"); 83 83 + } 84 84 + } 85 85 + // anything else on the debug channel is ignored 86 86 + } 87 87 + Ok(None) => break true, // client closed the connection 88 88 + Err(error) => { 89 89 + warn!(%error, "debug channel read failed"); 90 90 + break true; 91 91 + } 92 92 + }, 93 93 + } 94 94 + }; 95 95 + 96 96 + // if the client disconnected first, hang up the shell's process group so we 97 97 + // don't leak a detached session. (pty-process calls setsid in the child, so 98 98 + // it leads a new session and process group => pgid == pid.) 99 99 + if client_gone && let Some(pid) = pid { 100 100 + let _ = kill(Pid::from_raw(-(pid as i32)), Signal::SIGHUP); 101 101 + } 102 102 + 103 103 + let exit_code = match child.wait().await { 104 104 + Ok(status) => { 105 105 + use std::os::unix::process::ExitStatusExt; 106 106 + status 107 107 + .code() 108 108 + .or_else(|| status.signal().map(|signal| 128 + signal)) 109 109 + .unwrap_or(1) 110 110 + } 111 111 + Err(error) => { 112 112 + warn!(%error, "waiting on debug shell failed"); 113 113 + 1 114 114 + } 115 115 + }; 116 116 + 117 117 + let exit = Message { 118 118 + id: "pty".to_owned(), 119 119 + exec_exit: Some(v1::ExecExit { 120 120 + exit_code, 121 121 + error: String::new(), 122 122 + timed_out: false, 123 123 + }), 124 124 + ..Default::default() 125 125 + }; 126 126 + let _ = protocol::write_message(&mut conn_writer, &exit).await; 127 127 + info!(exit_code, "debug shell session ended"); 128 128 + Ok(()) 129 129 + } 130 130 + 131 131 + struct ResolvedUser { 132 132 + uid: u32, 133 133 + gid: u32, 134 134 + name: String, 135 135 + home: String, 136 136 + shell: String, 137 137 + } 138 138 + 139 139 + impl ResolvedUser { 140 140 + fn env(&self, term: &str) -> Vec<(String, String)> { 141 141 + let term = if term.is_empty() { 142 142 + "xterm-256color" 143 143 + } else { 144 144 + term 145 145 + }; 146 146 + vec![ 147 147 + ("TERM".to_owned(), term.to_owned()), 148 148 + ("HOME".to_owned(), self.home.clone()), 149 149 + ("USER".to_owned(), self.name.clone()), 150 150 + ("LOGNAME".to_owned(), self.name.clone()), 151 151 + ("SHELL".to_owned(), self.shell.clone()), 152 152 + ( 153 153 + "PATH".to_owned(), 154 154 + "/run/current-system/sw/bin:/usr/bin:/bin".to_owned(), 155 155 + ), 156 156 + ] 157 157 + } 158 158 + } 159 159 + 160 160 + fn resolve_user(name: &str) -> Result<ResolvedUser> { 161 161 + let user = User::from_name(name) 162 162 + .with_context(|| format!("lookup user {name:?}"))? 163 163 + .with_context(|| format!("debug shell user {name:?} not found"))?; 164 164 + if user.uid.as_raw() == 0 || user.gid.as_raw() == 0 { 165 165 + bail!("refusing to open a debug shell as privileged user {name:?}"); 166 166 + } 167 167 + let shell = user.shell.to_string_lossy().into_owned(); 168 168 + if shell.is_empty() { 169 169 + bail!("debug shell user {name:?} has no login shell set in the image"); 170 170 + } 171 171 + Ok(ResolvedUser { 172 172 + uid: user.uid.as_raw(), 173 173 + gid: user.gid.as_raw(), 174 174 + name: user.name, 175 175 + home: user.dir.to_string_lossy().into_owned(), 176 176 + shell, 177 177 + }) 178 178 + } 179 179 + 180 180 + fn clamp_tty_dim(value: u32) -> u16 { 181 181 + value.clamp(1, u16::MAX as u32) as u16 182 182 + }

+4 -1

shuttle/src/session.rs

View file Reviewed

··· 5 5 use crate::nix_config::{self, SYSTEMCTL_EXECUTABLE}; 6 6 use crate::on_payload; 7 7 use crate::protocol::{self, Message, v1}; 8 8 + use crate::pty; 8 9 use crate::{activation, command}; 9 10 use anyhow::{Context, Result, bail}; 10 11 use std::time::Duration; ··· 61 62 let read_result: Result<()> = loop { 62 63 tokio::select! { 63 64 read = protocol::read_message(&mut reader) => match read { 64 64 - Ok(Some(msg)) => spawn_message_task(&mut tasks, msg, &out_tx, uploader.clone()), 65 65 + Ok(Some(msg)) => spawn_message_task(&mut tasks, host_cid, msg, &out_tx, uploader.clone()), 65 66 Ok(None) => break Ok(()), 66 67 Err(error) => break Err(error).context("read message"), 67 68 }, ··· 84 85 85 86 fn spawn_message_task( 86 87 tasks: &mut JoinSet<()>, 88 88 + host_cid: u32, 87 89 msg: Message, 88 90 out_tx: &Sender<Message>, 89 91 uploader: Option<CacheUploadManager>, ··· 94 96 exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone())), 95 97 cache_drain => tasks.spawn(run_cache_drain(msg.id, cache_drain, out_tx.clone(), uploader)), 96 98 poweroff => tasks.spawn(run_poweroff(msg.id, poweroff, out_tx.clone())), 99 99 + open_debug_shell => tasks.spawn(pty::run(host_cid, open_debug_shell)), 97 100 }); 98 101 if handle.is_none() { 99 102 warn!(kind, "ignoring unsupported message");

+13

spindle/config/config.go

View file Reviewed

··· 94 94 95 95 AgingThreshold time.Duration `env:"AGING_THRESHOLD, default=30s"` 96 96 97 97 + DebugSSH DebugSSH `env:",prefix=DEBUG_SSH_"` 98 98 + 97 99 EnableCgroups bool `env:"ENABLE_CGROUPS, default=false"` 98 100 CgroupParent string `env:"CGROUP_PARENT, default=self"` 99 101 CgroupPidsMax int64 `env:"CGROUP_PIDS_MAX, default=4096"` ··· 106 108 CgroupIOWeight uint64 `env:"CGROUP_IO_WEIGHT, default=0"` 107 109 // memory.min that will get assigned to the supervisor (spindle itself) cgroup 108 110 CgroupSupervisorMemoryMinMiB int64 `env:"CGROUP_SUPERVISOR_MEMORY_MIN_MIB, default=512"` 111 111 + } 112 112 + 113 113 + type DebugSSH struct { 114 114 + Enabled bool `env:"ENABLED, default=false"` 115 115 + ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:2222"` 116 116 + JumpHost string `env:"JUMP_HOST"` 117 117 + Host string `env:"HOST"` 118 118 + // path to private key; if empty, spindle will generate one next to the db 119 119 + HostKeyPath string `env:"HOST_KEY_PATH"` 120 120 + // how long to keep a failed wf alive after failure, for sshing in 121 121 + GracePeriod time.Duration `env:"GRACE_PERIOD, default=5m"` 109 122 } 110 123 111 124 type NixCache struct {

shuttle/src/gen/file_descriptor_set.bin

View file Reviewed

This is a binary file and will not be displayed.

+230 -14

spindle/agentproto/gen/agent.pb.go

View file Reviewed

··· 780 780 return "" 781 781 } 782 782 783 783 + type OpenDebugShell struct { 784 784 + state protoimpl.MessageState `protogen:"open.v1"` 785 785 + VsockPort uint32 `protobuf:"varint,1,opt,name=vsock_port,json=vsockPort,proto3" json:"vsock_port,omitempty"` 786 786 + Term string `protobuf:"bytes,2,opt,name=term,proto3" json:"term,omitempty"` 787 787 + Rows uint32 `protobuf:"varint,3,opt,name=rows,proto3" json:"rows,omitempty"` 788 788 + Cols uint32 `protobuf:"varint,4,opt,name=cols,proto3" json:"cols,omitempty"` 789 789 + unknownFields protoimpl.UnknownFields 790 790 + sizeCache protoimpl.SizeCache 791 791 + } 792 792 + 793 793 + func (x *OpenDebugShell) Reset() { 794 794 + *x = OpenDebugShell{} 795 795 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] 796 796 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 797 797 + ms.StoreMessageInfo(mi) 798 798 + } 799 799 + 800 800 + func (x *OpenDebugShell) String() string { 801 801 + return protoimpl.X.MessageStringOf(x) 802 802 + } 803 803 + 804 804 + func (*OpenDebugShell) ProtoMessage() {} 805 805 + 806 806 + func (x *OpenDebugShell) ProtoReflect() protoreflect.Message { 807 807 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] 808 808 + if x != nil { 809 809 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 810 810 + if ms.LoadMessageInfo() == nil { 811 811 + ms.StoreMessageInfo(mi) 812 812 + } 813 813 + return ms 814 814 + } 815 815 + return mi.MessageOf(x) 816 816 + } 817 817 + 818 818 + // Deprecated: Use OpenDebugShell.ProtoReflect.Descriptor instead. 819 819 + func (*OpenDebugShell) Descriptor() ([]byte, []int) { 820 820 + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13} 821 821 + } 822 822 + 823 823 + func (x *OpenDebugShell) GetVsockPort() uint32 { 824 824 + if x != nil { 825 825 + return x.VsockPort 826 826 + } 827 827 + return 0 828 828 + } 829 829 + 830 830 + func (x *OpenDebugShell) GetTerm() string { 831 831 + if x != nil { 832 832 + return x.Term 833 833 + } 834 834 + return "" 835 835 + } 836 836 + 837 837 + func (x *OpenDebugShell) GetRows() uint32 { 838 838 + if x != nil { 839 839 + return x.Rows 840 840 + } 841 841 + return 0 842 842 + } 843 843 + 844 844 + func (x *OpenDebugShell) GetCols() uint32 { 845 845 + if x != nil { 846 846 + return x.Cols 847 847 + } 848 848 + return 0 849 849 + } 850 850 + 851 851 + // changes meaning based on who sends this: 852 852 + // guest->host is shell output, host->guest is keyboard input 853 853 + type PtyData struct { 854 854 + state protoimpl.MessageState `protogen:"open.v1"` 855 855 + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` 856 856 + unknownFields protoimpl.UnknownFields 857 857 + sizeCache protoimpl.SizeCache 858 858 + } 859 859 + 860 860 + func (x *PtyData) Reset() { 861 861 + *x = PtyData{} 862 862 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[14] 863 863 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 864 864 + ms.StoreMessageInfo(mi) 865 865 + } 866 866 + 867 867 + func (x *PtyData) String() string { 868 868 + return protoimpl.X.MessageStringOf(x) 869 869 + } 870 870 + 871 871 + func (*PtyData) ProtoMessage() {} 872 872 + 873 873 + func (x *PtyData) ProtoReflect() protoreflect.Message { 874 874 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[14] 875 875 + if x != nil { 876 876 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 877 877 + if ms.LoadMessageInfo() == nil { 878 878 + ms.StoreMessageInfo(mi) 879 879 + } 880 880 + return ms 881 881 + } 882 882 + return mi.MessageOf(x) 883 883 + } 884 884 + 885 885 + // Deprecated: Use PtyData.ProtoReflect.Descriptor instead. 886 886 + func (*PtyData) Descriptor() ([]byte, []int) { 887 887 + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{14} 888 888 + } 889 889 + 890 890 + func (x *PtyData) GetData() []byte { 891 891 + if x != nil { 892 892 + return x.Data 893 893 + } 894 894 + return nil 895 895 + } 896 896 + 897 897 + type PtyResize struct { 898 898 + state protoimpl.MessageState `protogen:"open.v1"` 899 899 + Rows uint32 `protobuf:"varint,1,opt,name=rows,proto3" json:"rows,omitempty"` 900 900 + Cols uint32 `protobuf:"varint,2,opt,name=cols,proto3" json:"cols,omitempty"` 901 901 + unknownFields protoimpl.UnknownFields 902 902 + sizeCache protoimpl.SizeCache 903 903 + } 904 904 + 905 905 + func (x *PtyResize) Reset() { 906 906 + *x = PtyResize{} 907 907 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[15] 908 908 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 909 909 + ms.StoreMessageInfo(mi) 910 910 + } 911 911 + 912 912 + func (x *PtyResize) String() string { 913 913 + return protoimpl.X.MessageStringOf(x) 914 914 + } 915 915 + 916 916 + func (*PtyResize) ProtoMessage() {} 917 917 + 918 918 + func (x *PtyResize) ProtoReflect() protoreflect.Message { 919 919 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[15] 920 920 + if x != nil { 921 921 + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 922 922 + if ms.LoadMessageInfo() == nil { 923 923 + ms.StoreMessageInfo(mi) 924 924 + } 925 925 + return ms 926 926 + } 927 927 + return mi.MessageOf(x) 928 928 + } 929 929 + 930 930 + // Deprecated: Use PtyResize.ProtoReflect.Descriptor instead. 931 931 + func (*PtyResize) Descriptor() ([]byte, []int) { 932 932 + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{15} 933 933 + } 934 934 + 935 935 + func (x *PtyResize) GetRows() uint32 { 936 936 + if x != nil { 937 937 + return x.Rows 938 938 + } 939 939 + return 0 940 940 + } 941 941 + 942 942 + func (x *PtyResize) GetCols() uint32 { 943 943 + if x != nil { 944 944 + return x.Cols 945 945 + } 946 946 + return 0 947 947 + } 948 948 + 783 949 type Message struct { 784 950 state protoimpl.MessageState `protogen:"open.v1"` 785 951 Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` ··· 796 962 CacheDrainResult *CacheDrainResult `protobuf:"bytes,12,opt,name=cache_drain_result,json=cacheDrainResult,proto3" json:"cache_drain_result,omitempty"` 797 963 Poweroff *Poweroff `protobuf:"bytes,13,opt,name=poweroff,proto3" json:"poweroff,omitempty"` 798 964 PoweroffResult *PoweroffResult `protobuf:"bytes,14,opt,name=poweroff_result,json=poweroffResult,proto3" json:"poweroff_result,omitempty"` 965 965 + OpenDebugShell *OpenDebugShell `protobuf:"bytes,15,opt,name=open_debug_shell,json=openDebugShell,proto3" json:"open_debug_shell,omitempty"` 966 966 + PtyData *PtyData `protobuf:"bytes,16,opt,name=pty_data,json=ptyData,proto3" json:"pty_data,omitempty"` 967 967 + PtyResize *PtyResize `protobuf:"bytes,17,opt,name=pty_resize,json=ptyResize,proto3" json:"pty_resize,omitempty"` 799 968 unknownFields protoimpl.UnknownFields 800 969 sizeCache protoimpl.SizeCache 801 970 } 802 971 803 972 func (x *Message) Reset() { 804 973 *x = Message{} 805 805 - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] 974 974 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[16] 806 975 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 807 976 ms.StoreMessageInfo(mi) 808 977 } ··· 814 983 func (*Message) ProtoMessage() {} 815 984 816 985 func (x *Message) ProtoReflect() protoreflect.Message { 817 817 - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] 986 986 + mi := &file_spindle_agent_v1_agent_proto_msgTypes[16] 818 987 if x != nil { 819 988 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 820 989 if ms.LoadMessageInfo() == nil { ··· 827 996 828 997 // Deprecated: Use Message.ProtoReflect.Descriptor instead. 829 998 func (*Message) Descriptor() ([]byte, []int) { 830 830 - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13} 999 999 + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{16} 831 1000 } 832 1001 833 1002 func (x *Message) GetId() string { ··· 928 1097 return nil 929 1098 } 930 1099 1100 1100 + func (x *Message) GetOpenDebugShell() *OpenDebugShell { 1101 1101 + if x != nil { 1102 1102 + return x.OpenDebugShell 1103 1103 + } 1104 1104 + return nil 1105 1105 + } 1106 1106 + 1107 1107 + func (x *Message) GetPtyData() *PtyData { 1108 1108 + if x != nil { 1109 1109 + return x.PtyData 1110 1110 + } 1111 1111 + return nil 1112 1112 + } 1113 1113 + 1114 1114 + func (x *Message) GetPtyResize() *PtyResize { 1115 1115 + if x != nil { 1116 1116 + return x.PtyResize 1117 1117 + } 1118 1118 + return nil 1119 1119 + } 1120 1120 + 931 1121 var File_spindle_agent_v1_agent_proto protoreflect.FileDescriptor 932 1122 933 1123 const file_spindle_agent_v1_agent_proto_rawDesc = "" + ··· 990 1180 "\n" + 991 1181 "\bPoweroff\"&\n" + 992 1182 "\x0ePoweroffResult\x12\x14\n" + 993 993 - "\x05error\x18\x01 \x01(\tR\x05error\"\xa8\b\n" + 1183 1183 + "\x05error\x18\x01 \x01(\tR\x05error\"k\n" + 1184 1184 + "\x0eOpenDebugShell\x12\x1d\n" + 1185 1185 + "\n" + 1186 1186 + "vsock_port\x18\x01 \x01(\rR\tvsockPort\x12\x12\n" + 1187 1187 + "\x04term\x18\x02 \x01(\tR\x04term\x12\x12\n" + 1188 1188 + "\x04rows\x18\x03 \x01(\rR\x04rows\x12\x12\n" + 1189 1189 + "\x04cols\x18\x04 \x01(\rR\x04cols\"\x1d\n" + 1190 1190 + "\aPtyData\x12\x12\n" + 1191 1191 + "\x04data\x18\x01 \x01(\fR\x04data\"3\n" + 1192 1192 + "\tPtyResize\x12\x12\n" + 1193 1193 + "\x04rows\x18\x01 \x01(\rR\x04rows\x12\x12\n" + 1194 1194 + "\x04cols\x18\x02 \x01(\rR\x04cols\"\x8e\n" + 1195 1195 + "\n" + 994 1196 "\aMessage\x12\x17\n" + 995 1197 "\x02id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x02id\x12-\n" + 996 1198 "\x05hello\x18\x02 \x01(\v2\x17.spindle.agent.v1.HelloR\x05hello\x12*\n" + ··· 1011 1213 "cacheDrain\x12P\n" + 1012 1214 "\x12cache_drain_result\x18\f \x01(\v2\".spindle.agent.v1.CacheDrainResultR\x10cacheDrainResult\x126\n" + 1013 1215 "\bpoweroff\x18\r \x01(\v2\x1a.spindle.agent.v1.PoweroffR\bpoweroff\x12I\n" + 1014 1014 - "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xb9\x01\xbaH\xb5\x01\"\xb2\x01\n" + 1216 1216 + "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult\x12J\n" + 1217 1217 + "\x10open_debug_shell\x18\x0f \x01(\v2 .spindle.agent.v1.OpenDebugShellR\x0eopenDebugShell\x124\n" + 1218 1218 + "\bpty_data\x18\x10 \x01(\v2\x19.spindle.agent.v1.PtyDataR\aptyData\x12:\n" + 1219 1219 + "\n" + 1220 1220 + "pty_resize\x18\x11 \x01(\v2\x1b.spindle.agent.v1.PtyResizeR\tptyResize:\xe1\x01\xbaH\xdd\x01\"\xda\x01\n" + 1015 1221 "\x05hello\n" + 1016 1222 "\x04init\n" + 1017 1223 "\n" + ··· 1025 1231 "\vcache_drain\n" + 1026 1232 "\x12cache_drain_result\n" + 1027 1233 "\bpoweroff\n" + 1028 1028 - "\x0fpoweroff_result\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" 1234 1234 + "\x0fpoweroff_result\n" + 1235 1235 + "\x10open_debug_shell\n" + 1236 1236 + "\bpty_data\n" + 1237 1237 + "\n" + 1238 1238 + "pty_resize\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" 1029 1239 1030 1240 var ( 1031 1241 file_spindle_agent_v1_agent_proto_rawDescOnce sync.Once ··· 1039 1249 return file_spindle_agent_v1_agent_proto_rawDescData 1040 1250 } 1041 1251 1042 1042 - var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 14) 1252 1252 + var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 17) 1043 1253 var file_spindle_agent_v1_agent_proto_goTypes = []any{ 1044 1254 (*Hello)(nil), // 0: spindle.agent.v1.Hello 1045 1255 (*Init)(nil), // 1: spindle.agent.v1.Init ··· 1054 1264 (*CacheDrainResult)(nil), // 10: spindle.agent.v1.CacheDrainResult 1055 1265 (*Poweroff)(nil), // 11: spindle.agent.v1.Poweroff 1056 1266 (*PoweroffResult)(nil), // 12: spindle.agent.v1.PoweroffResult 1057 1057 - (*Message)(nil), // 13: spindle.agent.v1.Message 1267 1267 + (*OpenDebugShell)(nil), // 13: spindle.agent.v1.OpenDebugShell 1268 1268 + (*PtyData)(nil), // 14: spindle.agent.v1.PtyData 1269 1269 + (*PtyResize)(nil), // 15: spindle.agent.v1.PtyResize 1270 1270 + (*Message)(nil), // 16: spindle.agent.v1.Message 1058 1271 } 1059 1272 var file_spindle_agent_v1_agent_proto_depIdxs = []int32{ 1060 1273 0, // 0: spindle.agent.v1.Message.hello:type_name -> spindle.agent.v1.Hello ··· 1070 1283 10, // 10: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult 1071 1284 11, // 11: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff 1072 1285 12, // 12: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult 1073 1073 - 13, // [13:13] is the sub-list for method output_type 1074 1074 - 13, // [13:13] is the sub-list for method input_type 1075 1075 - 13, // [13:13] is the sub-list for extension type_name 1076 1076 - 13, // [13:13] is the sub-list for extension extendee 1077 1077 - 0, // [0:13] is the sub-list for field type_name 1286 1286 + 13, // 13: spindle.agent.v1.Message.open_debug_shell:type_name -> spindle.agent.v1.OpenDebugShell 1287 1287 + 14, // 14: spindle.agent.v1.Message.pty_data:type_name -> spindle.agent.v1.PtyData 1288 1288 + 15, // 15: spindle.agent.v1.Message.pty_resize:type_name -> spindle.agent.v1.PtyResize 1289 1289 + 16, // [16:16] is the sub-list for method output_type 1290 1290 + 16, // [16:16] is the sub-list for method input_type 1291 1291 + 16, // [16:16] is the sub-list for extension type_name 1292 1292 + 16, // [16:16] is the sub-list for extension extendee 1293 1293 + 0, // [0:16] is the sub-list for field type_name 1078 1294 } 1079 1295 1080 1296 func init() { file_spindle_agent_v1_agent_proto_init() } ··· 1088 1304 GoPackagePath: reflect.TypeOf(x{}).PkgPath(), 1089 1305 RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_agent_v1_agent_proto_rawDesc), len(file_spindle_agent_v1_agent_proto_rawDesc)), 1090 1306 NumEnums: 0, 1091 1091 - NumMessages: 14, 1307 1307 + NumMessages: 17, 1092 1308 NumExtensions: 0, 1093 1309 NumServices: 0, 1094 1310 },

+31 -2

spindle/engines/microvm/README.md

View file Reviewed

··· 41 41 spindle expects, they should work. That is: 42 42 - a guest agent is present inside of the image and when that image boots it will 43 43 get started, 44 44 - - `spindle-workflow` user exists, 45 45 - - and the work directory is configured (`/workspace`). 44 44 + - the `spindle-workflow` user exists, is unprivileged (non-zero uid/gid), and has 45 45 + a usable login shell and home dir set in the image's passwd: workflow steps run 46 46 + as this user, and the debug shell (see below) launches its passwd shell as a 47 47 + login shell in its home dir. an unset or `nologin`/`false` shell breaks debug 48 48 + ssh, 49 49 + - and the work directory is configured (`/workspace`, with `/workspace/repo` as 50 50 + the per-step working dir). 46 51 47 52 ## Image discovery 48 53 ··· 234 239 never made it to the destination store. The guest still only ever sees the same 235 240 HTTP binary-cache upload protocol over vsock; it never gets direct access to 236 241 SSH credentials or the destination store itself. 242 242 + 243 243 + ### Debug ssh 244 244 + 245 245 + When a workflow fails, spindle can keep its microVM alive for a configured grace 246 246 + window (`MicroVMPipelines.SSH`) and print an `ssh` invocation so you can poke at 247 247 + the failed VM interactively. Spindle terminates the ssh connection itself and 248 248 + bridges a pty into the live guest over the agent's vsock; the guest stays 249 249 + keyless and never runs an ssh daemon. 250 250 + 251 251 + Access mirrors a git push: the ssh username is the job id, and the offered 252 252 + public key is sent to the job's repo knot (`sh.tangled.repo.checkPushAllowed`). 253 253 + The session is accepted only if that key is allowed to push to the job's repo. 254 254 + 255 255 + The shell is deliberately not configurable from either end. It always: 256 256 + - runs as the `spindle-workflow` user (the ssh username selects the *job*, not a 257 257 + unix user), 258 258 + - uses that user's login shell from the image's passwd, launched as a login 259 259 + shell (`-l`), and 260 260 + - starts in the dir where the repo was cloned to. 261 261 + 262 262 + The only things the client influences are the terminal type and window size 263 263 + (forwarded from the ssh pty request, and on resize). This relies on the image 264 264 + configuring `spindle-workflow` properly per the expectations above; in 265 265 + particular a missing or `nologin`/`false` won't work of course.

+13

spindle/engines/microvm/agent.go

View file Reviewed

··· 252 252 } 253 253 } 254 254 255 255 + func (s *AgentSession) OpenDebugShell(req *agentv1.OpenDebugShell) error { 256 256 + s.mu.Lock() 257 257 + defer s.mu.Unlock() 258 258 + 259 259 + if err := s.enc.Encode(&agentproto.Message{ 260 260 + Id: "debug-shell", 261 261 + OpenDebugShell: req, 262 262 + }); err != nil { 263 263 + return fmt.Errorf("send open_debug_shell: %w", err) 264 264 + } 265 265 + return nil 266 266 + } 267 267 + 255 268 func (s *AgentSession) Poweroff(ctx context.Context) error { 256 269 s.mu.Lock() 257 270 defer s.mu.Unlock()

+326

spindle/engines/microvm/debug.go

View file Reviewed

··· 1 1 + package microvm 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "encoding/hex" 6 6 + "errors" 7 7 + "fmt" 8 8 + "hash/fnv" 9 9 + "io" 10 10 + "log/slog" 11 11 + "net" 12 12 + "regexp" 13 13 + "strings" 14 14 + "sync" 15 15 + "time" 16 16 + 17 17 + "tangled.org/core/spindle/agentproto" 18 18 + agentv1 "tangled.org/core/spindle/agentproto/gen" 19 19 + "tangled.org/core/spindle/models" 20 20 + ) 21 21 + 22 22 + const debugAcceptTimeout = 15 * time.Second 23 23 + 24 24 + type debugTarget struct { 25 25 + cid uint32 26 26 + agent *AgentSession 27 27 + knot string 28 28 + repoDid string 29 29 + maxAliveAt time.Time 30 30 + connected chan struct{} // closed when the user first ssh's in, ending the grace window 31 31 + released chan struct{} // closed when the user exits the debug shell, to tear down early 32 32 + } 33 33 + 34 34 + var debugHandleSafe = regexp.MustCompile(`[^a-zA-Z0-9_.-]`) 35 35 + 36 36 + func newDebugHandle(wid models.WorkflowId) string { 37 37 + h := fnv.New32a() 38 38 + _, _ = io.WriteString(h, wid.String()) 39 39 + token := hex.EncodeToString(h.Sum(nil)) 40 40 + name := strings.Trim(debugHandleSafe.ReplaceAllString(wid.Name, "-"), "-") 41 41 + if name == "" { 42 42 + return token 43 43 + } 44 44 + return name + "-" + token 45 45 + } 46 46 + 47 47 + func (e *Engine) registerDebugTarget(wid models.WorkflowId, t debugTarget) { 48 48 + e.debugMu.Lock() 49 49 + defer e.debugMu.Unlock() 50 50 + e.debug[newDebugHandle(wid)] = t 51 51 + } 52 52 + 53 53 + func (e *Engine) unregisterDebugTarget(wid models.WorkflowId) { 54 54 + e.debugMu.Lock() 55 55 + defer e.debugMu.Unlock() 56 56 + delete(e.debug, newDebugHandle(wid)) 57 57 + } 58 58 + 59 59 + // ends the grace window so we hold the VM until the shell exits instead. 60 60 + func (e *Engine) markDebugConnected(handle string) { 61 61 + e.debugMu.Lock() 62 62 + defer e.debugMu.Unlock() 63 63 + if t, ok := e.debug[handle]; ok { 64 64 + select { 65 65 + case <-t.connected: // already signalled 66 66 + default: 67 67 + close(t.connected) 68 68 + } 69 69 + } 70 70 + } 71 71 + 72 72 + func (e *Engine) releaseDebugTarget(handle string) { 73 73 + e.debugMu.Lock() 74 74 + defer e.debugMu.Unlock() 75 75 + t, ok := e.debug[handle] 76 76 + if !ok { 77 77 + return 78 78 + } 79 79 + delete(e.debug, handle) 80 80 + close(t.released) 81 81 + } 82 82 + 83 83 + func (e *Engine) lookupDebugTarget(handle string) (debugTarget, bool) { 84 84 + e.debugMu.Lock() 85 85 + defer e.debugMu.Unlock() 86 86 + t, ok := e.debug[handle] 87 87 + return t, ok 88 88 + } 89 89 + 90 90 + func (e *Engine) RepoForJob(jobID string) (knot, repoDid string, ok bool) { 91 91 + t, found := e.lookupDebugTarget(jobID) 92 92 + if !found || t.knot == "" || t.repoDid == "" { 93 93 + return "", "", false 94 94 + } 95 95 + return t.knot, t.repoDid, true 96 96 + } 97 97 + 98 98 + func (e *Engine) OpenDebugSession(ctx context.Context, jobID, term string, rows, cols int) (*DebugSession, error) { 99 99 + target, ok := e.lookupDebugTarget(jobID) 100 100 + if !ok { 101 101 + return nil, fmt.Errorf("no live microVM for job %q", jobID) 102 102 + } 103 103 + 104 104 + ln, port, err := listenRandomVsockPort(ctx) 105 105 + if err != nil { 106 106 + return nil, fmt.Errorf("listen for debug shell: %w", err) 107 107 + } 108 108 + filtered := &cidFilteredVsockListener{Listener: ln, cid: target.cid, logger: e.l} 109 109 + 110 110 + if err := target.agent.OpenDebugShell(&agentv1.OpenDebugShell{ 111 111 + VsockPort: port, 112 112 + Term: term, 113 113 + Rows: clampDim(rows), 114 114 + Cols: clampDim(cols), 115 115 + }); err != nil { 116 116 + _ = ln.Close() 117 117 + return nil, fmt.Errorf("ask guest to open debug shell: %w", err) 118 118 + } 119 119 + 120 120 + conn, err := acceptWithTimeout(ctx, filtered, debugAcceptTimeout) 121 121 + if err != nil { 122 122 + _ = ln.Close() 123 123 + return nil, fmt.Errorf("accept debug shell connection: %w", err) 124 124 + } 125 125 + 126 126 + e.markDebugConnected(jobID) 127 127 + 128 128 + return newDebugSession(conn, ln, e.l), nil 129 129 + } 130 130 + 131 131 + func (e *Engine) writeDebugHint(wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) { 132 132 + if wfLogger == nil { 133 133 + wfLogger = models.NullLogger{} 134 134 + } 135 135 + step := Step{name: "Debug shell", kind: models.StepKindSystem} 136 136 + 137 137 + wfLogger.ControlWriter(idx, step, models.StepStatusStart).Write([]byte{0}) 138 138 + defer wfLogger.ControlWriter(idx, step, models.StepStatusEnd).Write([]byte{0}) 139 139 + 140 140 + ssh := e.cfg.MicroVMPipelines.DebugSSH 141 141 + grace := ssh.GracePeriod 142 142 + cmd := debugSSHCommand(ssh.ListenAddr, e.cfg.Server.Hostname, ssh.Host, ssh.JumpHost, newDebugHandle(wid)) 143 143 + out := wfLogger.DataWriter(idx, "stdout") 144 144 + fmt.Fprintf(out, "Workflow failed, connect within %s to debug until shell exit or workflow timeout:\n", grace) 145 145 + fmt.Fprintf(out, " %s\n", cmd) 146 146 + } 147 147 + 148 148 + func (e *Engine) maybeRetainForDebug(ctx context.Context, wid models.WorkflowId) { 149 149 + handle := newDebugHandle(wid) 150 150 + target, ok := e.lookupDebugTarget(handle) 151 151 + if !ok { 152 152 + return 153 153 + } 154 154 + 155 155 + grace := e.cfg.MicroVMPipelines.DebugSSH.GracePeriod 156 156 + e.l.Info("retaining failed microVM for debug", "workflow", wid, "grace", grace.String()) 157 157 + 158 158 + maxAlive := time.NewTimer(time.Until(target.maxAliveAt)) 159 159 + defer maxAlive.Stop() 160 160 + graceTimer := time.NewTimer(grace) 161 161 + defer graceTimer.Stop() 162 162 + 163 163 + // wait for the user to ssh in within the grace window 164 164 + select { 165 165 + case <-ctx.Done(): 166 166 + return 167 167 + case <-maxAlive.C: 168 168 + e.l.Info("debug retention hit max VM lifetime; tearing down microVM", "workflow", wid) 169 169 + return 170 170 + case <-graceTimer.C: 171 171 + e.l.Info("nobody ssh'd in within grace; tearing down microVM", "workflow", wid) 172 172 + return 173 173 + case <-target.connected: 174 174 + e.l.Info("debug shell connected; holding microVM until exit", "workflow", wid) 175 175 + } 176 176 + 177 177 + // connected: hold the VM until the user exits or it hits its max lifetime 178 178 + select { 179 179 + case <-ctx.Done(): 180 180 + case <-maxAlive.C: 181 181 + e.l.Info("debug session hit max VM lifetime; tearing down microVM", "workflow", wid) 182 182 + case <-target.released: 183 183 + e.l.Info("debug shell exited; tearing down microVM", "workflow", wid) 184 184 + } 185 185 + } 186 186 + 187 187 + func debugSSHCommand(listenAddr, hostname, debugHost, jumpHost, jobID string) string { 188 188 + host, port := hostname, "" 189 189 + if debugHost != "" { 190 190 + host = debugHost 191 191 + } 192 192 + if _, p, err := net.SplitHostPort(listenAddr); err == nil { 193 193 + port = p 194 194 + } 195 195 + args := []string{"ssh", "-tt"} 196 196 + if jumpHost != "" { 197 197 + args = append(args, "-J", jumpHost) 198 198 + } 199 199 + if port != "" && port != "22" { 200 200 + args = append(args, "-p", port) 201 201 + } 202 202 + args = append(args, jobID+"@"+host) 203 203 + return strings.Join(args, " ") 204 204 + } 205 205 + 206 206 + // bridges an interactive shell over the agentproto vsock. 207 207 + // Read to it gets the shell output from guest, Write sends the keyboard input from user. 208 208 + type DebugSession struct { 209 209 + conn net.Conn 210 210 + ln net.Listener 211 211 + enc *agentproto.Encoder 212 212 + dec *agentproto.Decoder 213 213 + l *slog.Logger 214 214 + 215 215 + out chan []byte 216 216 + leftover []byte 217 217 + exitCode int 218 218 + closeOne sync.Once 219 219 + } 220 220 + 221 221 + func newDebugSession(conn net.Conn, ln net.Listener, l *slog.Logger) *DebugSession { 222 222 + d := &DebugSession{ 223 223 + conn: conn, 224 224 + ln: ln, 225 225 + enc: agentproto.NewEncoder(conn), 226 226 + dec: agentproto.NewDecoder(conn), 227 227 + l: l, 228 228 + out: make(chan []byte, 16), 229 229 + } 230 230 + go d.readLoop() 231 231 + return d 232 232 + } 233 233 + 234 234 + func (d *DebugSession) readLoop() { 235 235 + defer close(d.out) 236 236 + for { 237 237 + msg, err := d.dec.Decode() 238 238 + if err != nil { 239 239 + if !errors.Is(err, io.EOF) { 240 240 + d.l.Debug("debug shell decode ended", "error", err) 241 241 + } 242 242 + return 243 243 + } 244 244 + if p := msg.PtyData; p != nil && len(p.Data) > 0 { 245 245 + d.out <- p.Data 246 246 + } else if p := msg.ExecExit; p != nil { 247 247 + d.exitCode = int(p.ExitCode) 248 248 + return 249 249 + } 250 250 + } 251 251 + } 252 252 + 253 253 + func (d *DebugSession) Read(p []byte) (int, error) { 254 254 + if len(d.leftover) == 0 { 255 255 + chunk, ok := <-d.out 256 256 + if !ok { 257 257 + return 0, io.EOF 258 258 + } 259 259 + d.leftover = chunk 260 260 + } 261 261 + n := copy(p, d.leftover) 262 262 + d.leftover = d.leftover[n:] 263 263 + return n, nil 264 264 + } 265 265 + 266 266 + func (d *DebugSession) Write(p []byte) (int, error) { 267 267 + if err := d.enc.Encode(&agentproto.Message{ 268 268 + Id: "pty", 269 269 + PtyData: &agentv1.PtyData{Data: append([]byte(nil), p...)}, 270 270 + }); err != nil { 271 271 + return 0, err 272 272 + } 273 273 + return len(p), nil 274 274 + } 275 275 + 276 276 + func (d *DebugSession) Resize(rows, cols int) error { 277 277 + return d.enc.Encode(&agentproto.Message{ 278 278 + Id: "pty", 279 279 + PtyResize: &agentv1.PtyResize{Rows: clampDim(rows), Cols: clampDim(cols)}, 280 280 + }) 281 281 + } 282 282 + 283 283 + func (d *DebugSession) ExitCode() int { return d.exitCode } 284 284 + 285 285 + func (d *DebugSession) Close() error { 286 286 + var err error 287 287 + d.closeOne.Do(func() { 288 288 + err = d.conn.Close() 289 289 + if d.ln != nil { 290 290 + _ = d.ln.Close() 291 291 + } 292 292 + }) 293 293 + return err 294 294 + } 295 295 + 296 296 + func acceptWithTimeout(ctx context.Context, ln net.Listener, timeout time.Duration) (net.Conn, error) { 297 297 + ctx, cancel := context.WithTimeout(ctx, timeout) 298 298 + defer cancel() 299 299 + 300 300 + type result struct { 301 301 + conn net.Conn 302 302 + err error 303 303 + } 304 304 + ch := make(chan result, 1) 305 305 + go func() { 306 306 + conn, err := ln.Accept() 307 307 + ch <- result{conn, err} 308 308 + }() 309 309 + 310 310 + select { 311 311 + case <-ctx.Done(): 312 312 + return nil, ctx.Err() 313 313 + case r := <-ch: 314 314 + return r.conn, r.err 315 315 + } 316 316 + } 317 317 + 318 318 + func clampDim(v int) uint32 { 319 319 + if v < 1 { 320 320 + return 1 321 321 + } 322 322 + if v > 65535 { 323 323 + return 65535 324 324 + } 325 325 + return uint32(v) 326 326 + }

+21

spindle/engines/microvm/debug_test.go

View file Reviewed

··· 1 1 + //go:build linux 2 2 + 3 3 + package microvm 4 4 + 5 5 + import "testing" 6 6 + 7 7 + func TestDebugSSHCommandUsesJumpHost(t *testing.T) { 8 8 + got := debugSSHCommand("0.0.0.0:2224", "executor-a.internal", "127.0.0.1", "spindle.example", "job-1") 9 9 + want := "ssh -tt -J spindle.example -p 2224 job-1@127.0.0.1" 10 10 + if got != want { 11 11 + t.Fatalf("debug ssh command = %q, want %q", got, want) 12 12 + } 13 13 + } 14 14 + 15 15 + func TestDebugSSHCommandUsesExecutorHostWithoutJump(t *testing.T) { 16 16 + got := debugSSHCommand("0.0.0.0:22", "executor-a.example", "", "", "job-1") 17 17 + want := "ssh -tt job-1@executor-a.example" 18 18 + if got != want { 19 19 + t.Fatalf("debug ssh command = %q, want %q", got, want) 20 20 + } 21 21 + }

+168

spindle/engines/microvm/debugssh.go

View file Reviewed

··· 1 1 + package microvm 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "crypto/ed25519" 6 6 + "crypto/rand" 7 7 + "encoding/pem" 8 8 + "fmt" 9 9 + "io" 10 10 + "net/http" 11 11 + "os" 12 12 + "path/filepath" 13 13 + "time" 14 14 + 15 15 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 16 16 + "github.com/gliderlabs/ssh" 17 17 + gossh "golang.org/x/crypto/ssh" 18 18 + "tangled.org/core/api/tangled" 19 19 + "tangled.org/core/hostutil" 20 20 + ) 21 21 + 22 22 + // debug ssh: terminates ssh here, bridges a pty into a live (failed) microVM 23 23 + // over the guest's agent conn. the guest stays keyless. access mirrors git 24 24 + // push: the offered key goes to the repo's knot 25 25 + // (sh.tangled.repo.checkPushAllowed) and is accepted only if it can push to the 26 26 + // job's repo. 27 27 + 28 28 + const debugAuthTimeout = 5 * time.Second 29 29 + 30 30 + func (e *Engine) serveDebugSSH(ctx context.Context) { 31 31 + dbg := e.cfg.MicroVMPipelines.DebugSSH 32 32 + addr := dbg.ListenAddr 33 33 + httpc := &http.Client{Timeout: debugAuthTimeout} 34 34 + 35 35 + srv := &ssh.Server{ 36 36 + Addr: addr, 37 37 + Handler: e.debugHandle, 38 38 + PublicKeyHandler: func(c ssh.Context, key ssh.PublicKey) bool { 39 39 + return e.checkDebugAuth(c, c.User(), key, httpc) 40 40 + }, 41 41 + } 42 42 + keyPath := e.cfg.MicroVMPipelines.DebugSSH.HostKeyPath 43 43 + if keyPath == "" { 44 44 + keyPath = filepath.Join(filepath.Dir(e.cfg.Server.DBPath), "debug_ssh_host_key") 45 45 + } 46 46 + if err := ensureDebugHostKey(keyPath); err != nil { 47 47 + e.l.Error("debug ssh: ensure host key", "path", keyPath, "err", err) 48 48 + return 49 49 + } 50 50 + if err := srv.SetOption(ssh.HostKeyFile(keyPath)); err != nil { 51 51 + e.l.Error("debug ssh: load host key", "path", keyPath, "err", err) 52 52 + return 53 53 + } 54 54 + 55 55 + go func() { 56 56 + <-ctx.Done() 57 57 + _ = srv.Close() 58 58 + }() 59 59 + 60 60 + e.l.Info("starting debug ssh server", "address", addr) 61 61 + if err := srv.ListenAndServe(); err != nil && err != ssh.ErrServerClosed { 62 62 + e.l.Error("debug ssh server stopped", "err", err) 63 63 + } 64 64 + } 65 65 + 66 66 + func ensureDebugHostKey(path string) error { 67 67 + if _, err := os.Stat(path); err == nil { 68 68 + return nil 69 69 + } else if !os.IsNotExist(err) { 70 70 + return fmt.Errorf("stat host key: %w", err) 71 71 + } 72 72 + 73 73 + _, priv, err := ed25519.GenerateKey(rand.Reader) 74 74 + if err != nil { 75 75 + return fmt.Errorf("generate host key: %w", err) 76 76 + } 77 77 + block, err := gossh.MarshalPrivateKey(priv, "") 78 78 + if err != nil { 79 79 + return fmt.Errorf("marshal host key: %w", err) 80 80 + } 81 81 + 82 82 + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { 83 83 + return fmt.Errorf("create host key dir: %w", err) 84 84 + } 85 85 + if err := os.WriteFile(path, pem.EncodeToMemory(block), 0o600); err != nil { 86 86 + return fmt.Errorf("write host key: %w", err) 87 87 + } 88 88 + return nil 89 89 + } 90 90 + 91 91 + func (e *Engine) debugHandle(sess ssh.Session) { 92 92 + ptyReq, winCh, isPty := sess.Pty() 93 93 + if !isPty { 94 94 + io.WriteString(sess.Stderr(), "error: no terminal allocated; use `ssh -t`\n") 95 95 + _ = sess.Exit(1) 96 96 + return 97 97 + } 98 98 + 99 99 + jobID := sess.User() 100 100 + l := e.l.With("component", "debugssh", "job", jobID) 101 101 + 102 102 + debug, err := e.OpenDebugSession(sess.Context(), jobID, ptyReq.Term, ptyReq.Window.Height, ptyReq.Window.Width) 103 103 + if err != nil { 104 104 + fmt.Fprintf(sess.Stderr(), "error: %v\n", err) 105 105 + _ = sess.Exit(1) 106 106 + return 107 107 + } 108 108 + defer debug.Close() 109 109 + l.Info("debug shell opened") 110 110 + 111 111 + go func() { 112 112 + for win := range winCh { 113 113 + if err := debug.Resize(win.Height, win.Width); err != nil { 114 114 + l.Debug("debug ssh resize failed", "error", err) 115 115 + } 116 116 + } 117 117 + }() 118 118 + 119 119 + // keyboard -> shell, runs until the client hangs up 120 120 + go func() { _, _ = io.Copy(debug, sess) }() 121 121 + // shell -> client, returns when the shell exits (Read hits EOF) 122 122 + _, _ = io.Copy(sess, debug) 123 123 + 124 124 + code := debug.ExitCode() 125 125 + l.Info("debug shell closed", "exitCode", code) 126 126 + // the user is done; let retention tear the VM down now instead of waiting 127 127 + // out the rest of the grace period 128 128 + e.releaseDebugTarget(jobID) 129 129 + _ = sess.Exit(code) 130 130 + } 131 131 + 132 132 + func (e *Engine) checkDebugAuth(ctx context.Context, jobID string, key ssh.PublicKey, httpc *http.Client) bool { 133 133 + l := e.l.With("component", "debugssh", "job", jobID, "keyType", key.Type()) 134 134 + 135 135 + knot, repoDid, ok := e.RepoForJob(jobID) 136 136 + if !ok { 137 137 + l.Warn("debug ssh: no live job / unknown repo") 138 138 + return false 139 139 + } 140 140 + 141 141 + host, noSSL, err := hostutil.ParseHostname(knot) 142 142 + if err != nil { 143 143 + l.Error("debug ssh: bad knot host", "knot", knot, "error", err) 144 144 + return false 145 145 + } 146 146 + scheme := "https" 147 147 + if noSSL { 148 148 + scheme = "http" 149 149 + } 150 150 + xc := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, host), Client: httpc} 151 151 + 152 152 + reqCtx, cancel := context.WithTimeout(ctx, debugAuthTimeout) 153 153 + defer cancel() 154 154 + 155 155 + out, err := tangled.RepoCheckPushAllowed(reqCtx, xc, string(gossh.MarshalAuthorizedKey(key)), repoDid) 156 156 + if err != nil { 157 157 + l.Error("debug ssh: push-allowed check failed", "knot", knot, "repo", repoDid, "error", err) 158 158 + return false 159 159 + } 160 160 + if !out.Allowed { 161 161 + l.Warn("debug ssh: key not allowed to push", "knot", knot, "repo", repoDid) 162 162 + return false 163 163 + } 164 164 + if out.Did != nil { 165 165 + l.Info("debug ssh: authorized", "did", *out.Did, "knot", knot, "repo", repoDid) 166 166 + } 167 167 + return true 168 168 + }

+28 -2

spindle/engines/microvm/engine.go

View file Reviewed

··· 58 58 59 59 cleanupMu sync.Mutex 60 60 cleanup map[string][]cleanupFunc 61 61 + 62 62 + debugMu sync.Mutex 63 63 + debug map[string]debugTarget 61 64 } 62 65 63 66 type Step struct { ··· 88 91 } 89 92 } 90 93 91 91 - return &Engine{ 94 94 + e := &Engine{ 92 95 l: l, 93 96 cfg: cfg, 94 97 db: d, ··· 97 100 budget: budget, 98 101 maxWorkflow: max, 99 102 cleanup: make(map[string][]cleanupFunc), 100 100 - }, nil 103 103 + debug: make(map[string]debugTarget), 104 104 + } 105 105 + 106 106 + if cfg.MicroVMPipelines.DebugSSH.Enabled && cfg.MicroVMPipelines.DebugSSH.ListenAddr != "" { 107 107 + go e.serveDebugSSH(ctx) 108 108 + } 109 109 + 110 110 + return e, nil 101 111 } 102 112 103 113 func (e *Engine) ensureAgentHub() (*agentHub, error) { ··· 313 323 return err 314 324 } 315 325 state.VM = vm 326 326 + state.StartedAt = time.Now() 316 327 317 328 category = "Failed to connect to agent" 318 329 ··· 337 348 return err 338 349 } 339 350 state.Agent = agentSession 351 351 + state.CID = cid 340 352 wf.Data = state 341 353 342 354 e.registerCleanup(wid, func(ctx context.Context) error { ··· 411 423 412 424 if exitCode != 0 { 413 425 e.l.Debug("step exited non-zero", "workflow", wid, "step", step.Name(), "exitCode", exitCode) 426 426 + if e.cfg.MicroVMPipelines.DebugSSH.Enabled { 427 427 + e.registerDebugTarget(wid, debugTarget{ 428 428 + cid: state.CID, 429 429 + agent: state.Agent, 430 430 + knot: w.Environment["TANGLED_REPO_KNOT"], 431 431 + repoDid: w.Environment["TANGLED_REPO_REPO_DID"], 432 432 + maxAliveAt: state.StartedAt.Add(e.WorkflowTimeout()), 433 433 + connected: make(chan struct{}), 434 434 + released: make(chan struct{}), 435 435 + }) 436 436 + e.writeDebugHint(wid, len(w.Steps), wfLogger) 437 437 + } 414 438 return fmt.Errorf("User step error: exited with code %d", exitCode) 415 439 } 416 440 return nil ··· 572 596 } 573 597 574 598 func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { 599 599 + e.maybeRetainForDebug(ctx, wid) 600 600 + 575 601 fns := e.drainCleanups(wid) 576 602 577 603 var cleanupErr error

+5

spindle/engines/microvm/vm.go

View file Reviewed

··· 169 169 CacheReadURLs []string 170 170 CacheTrustedPublicKeys []string 171 171 VM VMHandle 172 172 + CID uint32 172 173 Agent *AgentSession 173 174 ReadCache *ReadCacheProxy 174 175 UploadCache *UploadCacheProxy 175 176 DNSProxy *DNSProxy 176 177 WorkDir string 177 178 NixOSToplevelCache nixosToplevelCacheStore 179 179 + StartedAt time.Time // when the VM booted, for the max-lifetime cap 178 180 } 179 181 180 182 func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state *workflowState) error { 181 183 if state == nil { 182 184 return nil 183 185 } 186 186 + 187 187 + // stop advertising this VM for debug shells before we tear it down 188 188 + e.unregisterDebugTarget(wid) 184 189 185 190 ctx = context.WithoutCancel(ctx) 186 191

+38 -4

spindle/agentproto/spindle/agent/v1/agent.proto

View file Reviewed

··· 82 82 string error = 1; 83 83 } 84 84 85 85 + message OpenDebugShell { 86 86 + uint32 vsock_port = 1; 87 87 + string term = 2; 88 88 + uint32 rows = 3; 89 89 + uint32 cols = 4; 90 90 + } 91 91 + 92 92 + // changes meaning based on who sends this: 93 93 + // guest->host is shell output, host->guest is keyboard input 94 94 + message PtyData { 95 95 + bytes data = 1; 96 96 + } 97 97 + 98 98 + message PtyResize { 99 99 + uint32 rows = 1; 100 100 + uint32 cols = 2; 101 101 + } 102 102 + 85 103 message Message { 86 104 option (buf.validate.message).oneof = { 87 105 fields: [ 88 88 - "hello", "init", "exec_start", "exec_stdout", "exec_stderr", "exec_exit", 89 89 - "activate_config", "activate_config_result", "built_paths", "cache_drain", 90 90 - "cache_drain_result", "poweroff", "poweroff_result" 91 91 - ], 106 106 + "hello", 107 107 + "init", 108 108 + "exec_start", 109 109 + "exec_stdout", 110 110 + "exec_stderr", 111 111 + "exec_exit", 112 112 + "activate_config", 113 113 + "activate_config_result", 114 114 + "built_paths", 115 115 + "cache_drain", 116 116 + "cache_drain_result", 117 117 + "poweroff", 118 118 + "poweroff_result", 119 119 + "open_debug_shell", 120 120 + "pty_data", 121 121 + "pty_resize" 122 122 + ] 92 123 required: true 93 124 }; 94 125 ··· 107 138 CacheDrainResult cache_drain_result = 12; 108 139 Poweroff poweroff = 13; 109 140 PoweroffResult poweroff_result = 14; 141 141 + OpenDebugShell open_debug_shell = 15; 142 142 + PtyData pty_data = 16; 143 143 + PtyResize pty_resize = 17; 110 144 }

+33

shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs

View file Reviewed

··· 113 113 pub error: ::prost::alloc::string::String, 114 114 } 115 115 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] 116 116 + pub struct OpenDebugShell { 117 117 + #[prost(uint32, tag = "1")] 118 118 + pub vsock_port: u32, 119 119 + #[prost(string, tag = "2")] 120 120 + pub term: ::prost::alloc::string::String, 121 121 + #[prost(uint32, tag = "3")] 122 122 + pub rows: u32, 123 123 + /// debug shells always run as the spindle-workflow user, with that user's 124 124 + /// login shell, starting in its home dir. nothing here is client-specifiable. 125 125 + #[prost(uint32, tag = "4")] 126 126 + pub cols: u32, 127 127 + } 128 128 + /// changes meaning based on who sends this: 129 129 + /// guest->host is shell output, host->guest is keyboard input 130 130 + #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] 131 131 + pub struct PtyData { 132 132 + #[prost(bytes = "bytes", tag = "1")] 133 133 + pub data: ::prost::bytes::Bytes, 134 134 + } 135 135 + #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] 136 136 + pub struct PtyResize { 137 137 + #[prost(uint32, tag = "1")] 138 138 + pub rows: u32, 139 139 + #[prost(uint32, tag = "2")] 140 140 + pub cols: u32, 141 141 + } 142 142 + #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] 116 143 pub struct Message { 117 144 #[prost(string, tag = "1")] 118 145 pub id: ::prost::alloc::string::String, ··· 142 169 pub poweroff: ::core::option::Option<Poweroff>, 143 170 #[prost(message, optional, tag = "14")] 144 171 pub poweroff_result: ::core::option::Option<PoweroffResult>, 172 172 + #[prost(message, optional, tag = "15")] 173 173 + pub open_debug_shell: ::core::option::Option<OpenDebugShell>, 174 174 + #[prost(message, optional, tag = "16")] 175 175 + pub pty_data: ::core::option::Option<PtyData>, 176 176 + #[prost(message, optional, tag = "17")] 177 177 + pub pty_resize: ::core::option::Option<PtyResize>, 145 178 } 146 179 // @@protoc_insertion_point(module)