From afa37a6c692ef9cd1efabb60645226e5769d920e Mon Sep 17 00:00:00 2001 From: dawn Date: Tue, 23 Jun 2026 13:26:15 +0000 Subject: [PATCH] spindle/microvm: add ssh debug into failed job VMs Signed-off-by: dawn --- Cargo.lock | 11 +++++++++++ docker-compose.yml | 8 +++++--- shuttle/Cargo.toml | 3 ++- nix/microvm/base.nix | 1 + nix/modules/spindle.nix | 37 +++++++++++++++++++++++++++++++++++++ shuttle/src/command.rs | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------- shuttle/src/main.rs | 1 + shuttle/src/protocol.rs | 6 ++++++ shuttle/src/pty.rs | 182 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ shuttle/src/session.rs | 5 ++++- spindle/config/config.go | 11 +++++++++++ spindle/engine/engine.go | 17 +++-------------- shuttle/src/gen/file_descriptor_set.bin | 0 spindle/agentproto/gen/agent.pb.go | 244 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------- spindle/engines/microvm/README.md | 33 +++++++++++++++++++++++++++++++-- spindle/engines/microvm/agent.go | 13 +++++++++++++ spindle/engines/microvm/debug.go | 326 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/engines/microvm/debugssh.go | 171 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/engines/microvm/engine.go | 36 ++++++++++++++++++++++++++++++------ spindle/engines/microvm/vm.go | 5 +++++ spindle/agentproto/spindle/agent/v1/agent.proto | 42 ++++++++++++++++++++++++++++++++++++++---- shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs | 33 +++++++++++++++++++++++++++++++++ 22 file(s) changed, 1239 insertion(s)(+), 59 deletion(s)(-) diff --git a/Cargo.lock b/Cargo.lock --- a/Cargo.lock +++ b/Cargo.lock @@ -2975,6 +2975,16 @@ ] [[package]] +name = "pty-process" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266" +dependencies = [ + "rustix", + "tokio", +] + +[[package]] name = "quick_cache" version = "0.6.22" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -3680,6 +3690,7 @@ "prost", "prost-protovalidate", "prost-reflect", + "pty-process", "serde", "serde_json", "tempfile", diff --git a/docker-compose.yml b/docker-compose.yml --- a/docker-compose.yml +++ b/docker-compose.yml @@ -183,12 +183,12 @@ SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11240" SPINDLE_S3_LOG_BUCKET: "" SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS: "false" - # route guest nix substitution + uploads through the local ncps cache. - # ncps re-signs on serve with cache.local's key, so the guest trusts the - # matching public key below (no signing happens in spindle itself). SPINDLE_NIX_CACHE_READ_URLS: http://ncps:8501 SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS: cache.local:F7YqpMzuBdILYd/v+wMZN2YKxCzliXQyFmeezOxw7rU= SPINDLE_NIX_CACHE_UPLOAD_URL: http://ncps:8501/upload + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED: true + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2223 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD: 10m # these two are required for cgroups, uncomment if testing # privileged: true # cgroup: host @@ -203,6 +203,8 @@ security_opt: - label=disable - seccomp=unconfined + ports: + - "2223:2223" volumes: - spindle-data:/var/lib/spindle - spindle-logs:/var/log/spindle diff --git a/shuttle/Cargo.toml b/shuttle/Cargo.toml --- a/shuttle/Cargo.toml +++ b/shuttle/Cargo.toml @@ -8,11 +8,12 @@ [dependencies] anyhow = "1" base64 = "0.22" -nix = { version = "0.31", features = ["fs", "process", "reboot", "signal", "user"] } +nix = { version = "0.31", features = ["fs", "process", "reboot", "signal", "term", "user"] } prost = "0.14" prost-reflect = "0.16" prost-protovalidate = "0.3" once_cell = "1" +pty-process = { version = "0.5.3", features = ["async"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tempfile = "3" diff --git a/nix/microvm/base.nix b/nix/microvm/base.nix --- a/nix/microvm/base.nix +++ b/nix/microvm/base.nix @@ -199,6 +199,7 @@ group = "spindle-workflow"; home = "/workspace"; createHome = false; + shell = pkgs.bashInteractive; }; users.users.spindle-workflow.extraGroups = lib.mkIf config.virtualisation.docker.enable [ "docker" diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -263,6 +263,39 @@ ''; }; }; + + debugSsh = { + enable = mkOption { + type = types.bool; + default = false; + description = '' + Enable the debug ssh server that lets authorized users ssh into a + failed microVM to debug it. + ''; + }; + listenAddr = mkOption { + type = types.str; + default = "0.0.0.0:2222"; + example = "0.0.0.0:2225"; + description = "Address for the debug ssh server to listen on."; + }; + hostKeyPath = mkOption { + type = with types; nullOr path; + default = null; + example = "/var/lib/spindle/debug_ssh_host_key"; + description = '' + Path to the ssh host key for the debug server. If null, one is generated + once and persisted next to the spindle db. + ''; + }; + gracePeriod = mkOption { + type = types.str; + default = "5m"; + description = '' + How long a failed workflow's microVM is kept alive for the user to ssh in. + ''; + }; + }; }; nixCache = { @@ -385,6 +418,10 @@ "SPINDLE_MICROVM_PIPELINES_CGROUP_PIDS_MAX=${toString cfg.pipelines.microvm.cgroup.pidsMax}" "SPINDLE_MICROVM_PIPELINES_CGROUP_SWAP_MAX_MIB=${toString cfg.pipelines.microvm.cgroup.swapMaxMiB}" "SPINDLE_MICROVM_PIPELINES_CGROUP_SUPERVISOR_MEMORY_MIN_MIB=${toString cfg.pipelines.microvm.cgroup.supervisorMinMiB}" + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED=${lib.boolToString cfg.pipelines.microvm.debugSsh.enable}" + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR=${cfg.pipelines.microvm.debugSsh.listenAddr}" + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST_KEY_PATH=${optionalString (cfg.pipelines.microvm.debugSsh.hostKeyPath != null) (toString cfg.pipelines.microvm.debugSsh.hostKeyPath)}" + "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD=${cfg.pipelines.microvm.debugSsh.gracePeriod}" "SPINDLE_NIX_CACHE_READ_URLS=${concatStringsSep "," cfg.pipelines.nixCache.readUrls}" "SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS=${concatStringsSep "," cfg.pipelines.nixCache.trustedPublicKeys}" "SPINDLE_NIX_CACHE_UPLOAD_URL=${cfg.pipelines.nixCache.uploadUrl}" diff --git a/shuttle/src/command.rs b/shuttle/src/command.rs --- a/shuttle/src/command.rs +++ b/shuttle/src/command.rs @@ -1,7 +1,8 @@ use anyhow::{Context, Result}; use nix::sys::signal::{Signal, kill}; use nix::unistd::{Gid, Pid, Uid, User, getgrouplist, setgid, setgroups, setuid}; -use std::ffi::{CString, OsStr, OsString}; +use pty_process::{Command as PtyCommand, OwnedReadPty, OwnedWritePty, Size}; +use std::ffi::{CString, OsString}; use std::io; use std::os::unix::process::ExitStatusExt; use std::path::PathBuf; @@ -196,16 +197,7 @@ // group won't actually let it access the sock. // https://github.com/rust-lang/rust/issues/90747 if let (Some(uid), Some(gid)) = (spec.uid, spec.gid) { - let username = User::from_uid(Uid::from_raw(uid)) - .ok() - .flatten() - .map(|u| u.name) - .with_context(|| format!("lookup passwd entry for uid {uid}"))?; - let cname = CString::new(username) - .with_context(|| format!("username for uid {uid} contained a null byte"))?; - // resolve groups beforehand so we don't have to read /etc/group in the pre_exec - let groups = - getgrouplist(&cname, Gid::from_raw(gid)).context("resolve supplementary groups")?; + let groups = resolve_supplementary_groups(uid, gid)?; // SAFETY: pre_exec runs between fork and execve in the child. // we only call async-signal-safe syscalls and we don't touch any // shared state, no allocator, no mutexes, no globals. @@ -223,7 +215,65 @@ cmd.process_group(0); cmd.spawn() - .with_context(|| format!("spawn {}", display_os(&spec.program))) + .with_context(|| format!("spawn {:?}", &spec.program)) +} + +// resolve the supplementary group list up front so the pre_exec hook never has +// to read /etc/group (which is not async-signal-safe) between fork and exec. +fn resolve_supplementary_groups(uid: u32, gid: u32) -> Result> { + let username = User::from_uid(Uid::from_raw(uid)) + .ok() + .flatten() + .map(|u| u.name) + .with_context(|| format!("lookup passwd entry for uid {uid}"))?; + let cname = CString::new(username) + .with_context(|| format!("username for uid {uid} contained a null byte"))?; + getgrouplist(&cname, Gid::from_raw(gid)).context("resolve supplementary groups") +} + +pub fn spawn_pty(spec: Spec, rows: u16, cols: u16) -> Result<(OwnedReadPty, OwnedWritePty, Child)> { + let (pty, pts) = pty_process::open().context("open pty")?; + pty.resize(Size::new(rows, cols)).context("set pty size")?; + + let mut cmd = PtyCommand::new(&spec.program) + .args(&spec.args) + .envs(spec.env.iter().map(|(key, value)| (key, value))); + if let Some(cwd) = &spec.cwd { + cmd = cmd.current_dir(cwd); + } + + // drop privileges in the child. this RELIES on pty-process composing our + // pre_exec hook *after* its own session setup: it wraps us as `move || { + // session_leader()?; ours()?; }`, so setsid + TIOCSCTTY run first (while + // still privileged) and only then do we drop to the workflow user. that + // ordering is what we want and we depend on it. if pty-process ever ran our + // hook first, the session setup would happen post-drop. (it'd likely still + // work, since setsid/TIOCSCTTY on our own pty need no privilege, but it is + // not the behaviour we're assuming here) + // don't use .uid()/.gid() here, they clear supplementary groups (see L195). + if let (Some(uid), Some(gid)) = (spec.uid, spec.gid) { + let groups = resolve_supplementary_groups(uid, gid)?; + // SAFETY: pre_exec runs between fork and execve in the child. every call + // below is async-signal-safe and touches no shared state. + cmd = unsafe { + cmd.pre_exec(move || { + setgroups(&groups).map_err(io::Error::from)?; + setgid(Gid::from_raw(gid)).map_err(io::Error::from)?; + setuid(Uid::from_raw(uid)).map_err(io::Error::from)?; + Ok(()) + }) + }; + } + + // spawn consumes the slave (dup'd onto the child's 0/1/2 and then closed in + // the parent), so the master reports EOF once the shell and all its children + // have exited. + let child = cmd + .spawn(pts) + .with_context(|| format!("spawn pty shell {:?}", &spec.program))?; + + let (reader, writer) = pty.into_split(); + Ok((reader, writer, child)) } async fn wait_child(child: &mut Child, timeout: Option) -> ExitResult { @@ -293,6 +343,41 @@ }) } -fn display_os(value: &OsStr) -> String { - value.to_string_lossy().into_owned() +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncReadExt; + + #[tokio::test] + async fn pty_runs_a_shell_and_reports_exit() { + let spec = Spec::new("/bin/sh") + .arg("-c") + .arg("printf 'hello pty'; exit 7"); + let (mut reader, _writer, mut child) = spawn_pty(spec, 24, 80).expect("spawn pty"); + + let mut output = Vec::new(); + let mut chunk = [0u8; 1024]; + loop { + match reader.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => output.extend_from_slice(&chunk[..n]), + // linux signals slave-closed with EIO rather than EOF + Err(error) if error.raw_os_error() == Some(nix::libc::EIO) => break, + Err(error) => panic!("read pty master: {error}"), + } + } + + let status = child.wait().await.expect("wait child"); + let text = String::from_utf8_lossy(&output); + assert!(text.contains("hello pty"), "unexpected output: {text:?}"); + assert_eq!(status.code(), Some(7)); + } + + #[tokio::test] + async fn pty_resize_succeeds() { + let spec = Spec::new("/bin/sh").arg("-c").arg("sleep 0.2"); + let (_reader, writer, mut child) = spawn_pty(spec, 24, 80).expect("spawn pty"); + writer.resize(Size::new(40, 120)).expect("resize"); + let _ = child.wait().await; + } } diff --git a/shuttle/src/main.rs b/shuttle/src/main.rs --- a/shuttle/src/main.rs +++ b/shuttle/src/main.rs @@ -9,6 +9,7 @@ mod logging; mod nix_config; mod protocol; +mod pty; mod session; use std::env; diff --git a/shuttle/src/protocol.rs b/shuttle/src/protocol.rs --- a/shuttle/src/protocol.rs +++ b/shuttle/src/protocol.rs @@ -43,6 +43,9 @@ CacheDrainResult, Poweroff, PoweroffResult, + OpenDebugShell, + PtyData, + PtyResize, Message, ); @@ -79,6 +82,9 @@ cache_drain_result => "cache_drain_result", poweroff => "poweroff", poweroff_result => "poweroff_result", + open_debug_shell => "open_debug_shell", + pty_data => "pty_data", + pty_resize => "pty_resize", }) .unwrap_or_else(|| unreachable!("validated message has no payload")) } diff --git a/shuttle/src/pty.rs b/shuttle/src/pty.rs new file mode 100644 --- /dev/null +++ b/shuttle/src/pty.rs @@ -0,0 +1,182 @@ +use std::path::Path; + +use crate::command::{self, Spec}; +use crate::protocol::{self, Message, v1}; +use anyhow::{Context, Result, bail}; +use nix::sys::signal::{Signal, kill}; +use nix::unistd::{Pid, User}; +use pty_process::Size; +use tokio::io::{AsyncReadExt, BufReader}; +use tokio_vsock::{VsockAddr, VsockStream}; +use tracing::{info, warn}; + +const WF_USER: &str = "spindle-workflow"; +const READ_CHUNK: usize = 32 * 1024; + +pub async fn run(host_cid: u32, open: v1::OpenDebugShell) { + if let Err(error) = serve(host_cid, open).await { + warn!(%error, "debug shell session failed"); + } +} + +async fn serve(host_cid: u32, open: v1::OpenDebugShell) -> Result<()> { + let conn = VsockStream::connect(VsockAddr::new(host_cid, open.vsock_port)) + .await + .with_context(|| format!("dial host debug vsock port {}", open.vsock_port))?; + info!(port = open.vsock_port, "debug shell connected"); + + let user = resolve_user(WF_USER)?; + + let rows = clamp_tty_dim(open.rows); + let cols = clamp_tty_dim(open.cols); + + let spec = Spec::new(&user.shell) + .arg("-l") + .envs(user.env(&open.term)) + .run_as(user.uid, user.gid) + .cwd(Path::new(&user.home).join("repo")); // /workflow/repo + + let (mut pty_reader, mut pty_writer, mut child) = + command::spawn_pty(spec, rows, cols).context("spawn pty shell")?; + let pid = child.id(); + + let (conn_reader, conn_writer) = tokio::io::split(conn); + let mut conn_reader = BufReader::new(conn_reader); + let mut conn_writer = conn_writer; + + let mut buf = vec![0u8; READ_CHUNK]; + let client_gone = loop { + tokio::select! { + read = pty_reader.read(&mut buf) => match read { + Ok(0) => break false, // shell exited + Ok(n) => { + let msg = Message { + id: "pty".to_owned(), + pty_data: Some(v1::PtyData { data: buf[..n].to_vec().into() }), + ..Default::default() + }; + if protocol::write_message(&mut conn_writer, &msg).await.is_err() { + break true; + } + } + Err(error) => { + // linux returns EIO (not a clean EOF) on the master once the + // slave side is fully closed, so treat that as the shell + // exiting normally rather than a real read failure. + if error.raw_os_error() != Some(nix::libc::EIO) { + warn!(%error, "pty master read failed"); + } + break false; + } + }, + incoming = protocol::read_message(&mut conn_reader) => match incoming { + Ok(Some(msg)) => { + if let Some(data) = msg.pty_data { + use tokio::io::AsyncWriteExt; + if pty_writer.write_all(&data.data).await.is_err() { + break false; + } + } else if let Some(resize) = msg.pty_resize { + let size = Size::new(clamp_tty_dim(resize.rows), clamp_tty_dim(resize.cols)); + if let Err(error) = pty_writer.resize(size) { + warn!(%error, "pty resize failed"); + } + } + // anything else on the debug channel is ignored + } + Ok(None) => break true, // client closed the connection + Err(error) => { + warn!(%error, "debug channel read failed"); + break true; + } + }, + } + }; + + // if the client disconnected first, hang up the shell's process group so we + // don't leak a detached session. (pty-process calls setsid in the child, so + // it leads a new session and process group => pgid == pid.) + if client_gone && let Some(pid) = pid { + let _ = kill(Pid::from_raw(-(pid as i32)), Signal::SIGHUP); + } + + let exit_code = match child.wait().await { + Ok(status) => { + use std::os::unix::process::ExitStatusExt; + status + .code() + .or_else(|| status.signal().map(|signal| 128 + signal)) + .unwrap_or(1) + } + Err(error) => { + warn!(%error, "waiting on debug shell failed"); + 1 + } + }; + + let exit = Message { + id: "pty".to_owned(), + exec_exit: Some(v1::ExecExit { + exit_code, + error: String::new(), + timed_out: false, + }), + ..Default::default() + }; + let _ = protocol::write_message(&mut conn_writer, &exit).await; + info!(exit_code, "debug shell session ended"); + Ok(()) +} + +struct ResolvedUser { + uid: u32, + gid: u32, + name: String, + home: String, + shell: String, +} + +impl ResolvedUser { + fn env(&self, term: &str) -> Vec<(String, String)> { + let term = if term.is_empty() { + "xterm-256color" + } else { + term + }; + vec![ + ("TERM".to_owned(), term.to_owned()), + ("HOME".to_owned(), self.home.clone()), + ("USER".to_owned(), self.name.clone()), + ("LOGNAME".to_owned(), self.name.clone()), + ("SHELL".to_owned(), self.shell.clone()), + ( + "PATH".to_owned(), + "/run/current-system/sw/bin:/usr/bin:/bin".to_owned(), + ), + ] + } +} + +fn resolve_user(name: &str) -> Result { + let user = User::from_name(name) + .with_context(|| format!("lookup user {name:?}"))? + .with_context(|| format!("debug shell user {name:?} not found"))?; + if user.uid.as_raw() == 0 || user.gid.as_raw() == 0 { + bail!("refusing to open a debug shell as privileged user {name:?}"); + } + let shell = user.shell.to_string_lossy().into_owned(); + if shell.is_empty() { + bail!("debug shell user {name:?} has no login shell set in the image"); + } + Ok(ResolvedUser { + uid: user.uid.as_raw(), + gid: user.gid.as_raw(), + name: user.name, + home: user.dir.to_string_lossy().into_owned(), + shell, + }) +} + +fn clamp_tty_dim(value: u32) -> u16 { + value.clamp(1, u16::MAX as u32) as u16 +} diff --git a/shuttle/src/session.rs b/shuttle/src/session.rs --- a/shuttle/src/session.rs +++ b/shuttle/src/session.rs @@ -5,6 +5,7 @@ use crate::nix_config::{self, SYSTEMCTL_EXECUTABLE}; use crate::on_payload; use crate::protocol::{self, Message, v1}; +use crate::pty; use crate::{activation, command}; use anyhow::{Context, Result, bail}; use std::time::Duration; @@ -61,7 +62,7 @@ let read_result: Result<()> = loop { tokio::select! { read = protocol::read_message(&mut reader) => match read { - Ok(Some(msg)) => spawn_message_task(&mut tasks, msg, &out_tx, uploader.clone()), + Ok(Some(msg)) => spawn_message_task(&mut tasks, host_cid, msg, &out_tx, uploader.clone()), Ok(None) => break Ok(()), Err(error) => break Err(error).context("read message"), }, @@ -84,6 +85,7 @@ fn spawn_message_task( tasks: &mut JoinSet<()>, + host_cid: u32, msg: Message, out_tx: &Sender, uploader: Option, @@ -94,6 +96,7 @@ exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone())), cache_drain => tasks.spawn(run_cache_drain(msg.id, cache_drain, out_tx.clone(), uploader)), poweroff => tasks.spawn(run_poweroff(msg.id, poweroff, out_tx.clone())), + open_debug_shell => tasks.spawn(pty::run(host_cid, open_debug_shell)), }); if handle.is_none() { warn!(kind, "ignoring unsupported message"); diff --git a/spindle/config/config.go b/spindle/config/config.go --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -79,12 +79,23 @@ AgingThreshold time.Duration `env:"AGING_THRESHOLD, default=30s"` + DebugSSH DebugSSH `env:",prefix=DEBUG_SSH_"` + EnableCgroups bool `env:"ENABLE_CGROUPS, default=false"` CgroupParent string `env:"CGROUP_PARENT, default=self"` CgroupPidsMax int64 `env:"CGROUP_PIDS_MAX, default=4096"` CgroupSwapMaxMiB *int64 `env:"CGROUP_SWAP_MAX_MIB"` // memory.min that will get assigned to the supervisor (spindle itself) cgroup CgroupSupervisorMemoryMinMiB int64 `env:"CGROUP_SUPERVISOR_MEMORY_MIN_MIB, default=512"` +} + +type DebugSSH struct { + Enabled bool `env:"ENABLED, default=false"` + ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:2222"` + // path to private key; if empty, spindle will generate one next to the db + HostKeyPath string `env:"HOST_KEY_PATH"` + // how long to keep a failed wf alive after failure, for sshing in + GracePeriod time.Duration `env:"GRACE_PERIOD, default=5m"` } type NixCache struct { diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -20,10 +20,6 @@ ErrWorkflowFailed = errors.New("workflow failed") ) -type workflowFinalizer interface { - FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error -} - func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) { l.Info("starting all workflows in parallel", "pipeline", pipelineId) @@ -116,6 +112,9 @@ } return } + // don't put this after the workflowTimeout deadline assignment + // below. engines that implement "ssh-after-fail" rely on the + // unbounded ctx for retaining the workflow after it fails. defer eng.DestroyWorkflow(ctx, wid) ctx, cancel := context.WithTimeout(ctx, workflowTimeout) @@ -149,16 +148,6 @@ if dbErr != nil { l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) } - } - return - } - } - - if finalizer, ok := eng.(workflowFinalizer); ok { - if err := finalizer.FinalizeWorkflow(ctx, wid, &w, wfLogger); err != nil { - dbErr := db.StatusFailed(wid, err.Error(), -1, n) - if dbErr != nil { - l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) } return } diff --git a/shuttle/src/gen/file_descriptor_set.bin b/shuttle/src/gen/file_descriptor_set.bin --- a/shuttle/src/gen/file_descriptor_set.bin +++ b/shuttle/src/gen/file_descriptor_set.bin diff --git a/spindle/agentproto/gen/agent.pb.go b/spindle/agentproto/gen/agent.pb.go --- a/spindle/agentproto/gen/agent.pb.go +++ b/spindle/agentproto/gen/agent.pb.go @@ -780,6 +780,172 @@ return "" } +type OpenDebugShell struct { + state protoimpl.MessageState `protogen:"open.v1"` + VsockPort uint32 `protobuf:"varint,1,opt,name=vsock_port,json=vsockPort,proto3" json:"vsock_port,omitempty"` + Term string `protobuf:"bytes,2,opt,name=term,proto3" json:"term,omitempty"` + Rows uint32 `protobuf:"varint,3,opt,name=rows,proto3" json:"rows,omitempty"` + Cols uint32 `protobuf:"varint,4,opt,name=cols,proto3" json:"cols,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *OpenDebugShell) Reset() { + *x = OpenDebugShell{} + mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *OpenDebugShell) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*OpenDebugShell) ProtoMessage() {} + +func (x *OpenDebugShell) ProtoReflect() protoreflect.Message { + mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use OpenDebugShell.ProtoReflect.Descriptor instead. +func (*OpenDebugShell) Descriptor() ([]byte, []int) { + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13} +} + +func (x *OpenDebugShell) GetVsockPort() uint32 { + if x != nil { + return x.VsockPort + } + return 0 +} + +func (x *OpenDebugShell) GetTerm() string { + if x != nil { + return x.Term + } + return "" +} + +func (x *OpenDebugShell) GetRows() uint32 { + if x != nil { + return x.Rows + } + return 0 +} + +func (x *OpenDebugShell) GetCols() uint32 { + if x != nil { + return x.Cols + } + return 0 +} + +// changes meaning based on who sends this: +// guest->host is shell output, host->guest is keyboard input +type PtyData struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PtyData) Reset() { + *x = PtyData{} + mi := &file_spindle_agent_v1_agent_proto_msgTypes[14] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PtyData) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PtyData) ProtoMessage() {} + +func (x *PtyData) ProtoReflect() protoreflect.Message { + mi := &file_spindle_agent_v1_agent_proto_msgTypes[14] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PtyData.ProtoReflect.Descriptor instead. +func (*PtyData) Descriptor() ([]byte, []int) { + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{14} +} + +func (x *PtyData) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + +type PtyResize struct { + state protoimpl.MessageState `protogen:"open.v1"` + Rows uint32 `protobuf:"varint,1,opt,name=rows,proto3" json:"rows,omitempty"` + Cols uint32 `protobuf:"varint,2,opt,name=cols,proto3" json:"cols,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PtyResize) Reset() { + *x = PtyResize{} + mi := &file_spindle_agent_v1_agent_proto_msgTypes[15] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PtyResize) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PtyResize) ProtoMessage() {} + +func (x *PtyResize) ProtoReflect() protoreflect.Message { + mi := &file_spindle_agent_v1_agent_proto_msgTypes[15] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PtyResize.ProtoReflect.Descriptor instead. +func (*PtyResize) Descriptor() ([]byte, []int) { + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{15} +} + +func (x *PtyResize) GetRows() uint32 { + if x != nil { + return x.Rows + } + return 0 +} + +func (x *PtyResize) GetCols() uint32 { + if x != nil { + return x.Cols + } + return 0 +} + type Message struct { state protoimpl.MessageState `protogen:"open.v1"` Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` @@ -796,13 +962,16 @@ CacheDrainResult *CacheDrainResult `protobuf:"bytes,12,opt,name=cache_drain_result,json=cacheDrainResult,proto3" json:"cache_drain_result,omitempty"` Poweroff *Poweroff `protobuf:"bytes,13,opt,name=poweroff,proto3" json:"poweroff,omitempty"` PoweroffResult *PoweroffResult `protobuf:"bytes,14,opt,name=poweroff_result,json=poweroffResult,proto3" json:"poweroff_result,omitempty"` + OpenDebugShell *OpenDebugShell `protobuf:"bytes,15,opt,name=open_debug_shell,json=openDebugShell,proto3" json:"open_debug_shell,omitempty"` + PtyData *PtyData `protobuf:"bytes,16,opt,name=pty_data,json=ptyData,proto3" json:"pty_data,omitempty"` + PtyResize *PtyResize `protobuf:"bytes,17,opt,name=pty_resize,json=ptyResize,proto3" json:"pty_resize,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *Message) Reset() { *x = Message{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -814,7 +983,7 @@ func (*Message) ProtoMessage() {} func (x *Message) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -827,7 +996,7 @@ // Deprecated: Use Message.ProtoReflect.Descriptor instead. func (*Message) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{16} } func (x *Message) GetId() string { @@ -928,6 +1097,27 @@ return nil } +func (x *Message) GetOpenDebugShell() *OpenDebugShell { + if x != nil { + return x.OpenDebugShell + } + return nil +} + +func (x *Message) GetPtyData() *PtyData { + if x != nil { + return x.PtyData + } + return nil +} + +func (x *Message) GetPtyResize() *PtyResize { + if x != nil { + return x.PtyResize + } + return nil +} + var File_spindle_agent_v1_agent_proto protoreflect.FileDescriptor const file_spindle_agent_v1_agent_proto_rawDesc = "" + @@ -990,7 +1180,19 @@ "\n" + "\bPoweroff\"&\n" + "\x0ePoweroffResult\x12\x14\n" + - "\x05error\x18\x01 \x01(\tR\x05error\"\xa8\b\n" + + "\x05error\x18\x01 \x01(\tR\x05error\"k\n" + + "\x0eOpenDebugShell\x12\x1d\n" + + "\n" + + "vsock_port\x18\x01 \x01(\rR\tvsockPort\x12\x12\n" + + "\x04term\x18\x02 \x01(\tR\x04term\x12\x12\n" + + "\x04rows\x18\x03 \x01(\rR\x04rows\x12\x12\n" + + "\x04cols\x18\x04 \x01(\rR\x04cols\"\x1d\n" + + "\aPtyData\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\"3\n" + + "\tPtyResize\x12\x12\n" + + "\x04rows\x18\x01 \x01(\rR\x04rows\x12\x12\n" + + "\x04cols\x18\x02 \x01(\rR\x04cols\"\x8e\n" + + "\n" + "\aMessage\x12\x17\n" + "\x02id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x02id\x12-\n" + "\x05hello\x18\x02 \x01(\v2\x17.spindle.agent.v1.HelloR\x05hello\x12*\n" + @@ -1011,7 +1213,11 @@ "cacheDrain\x12P\n" + "\x12cache_drain_result\x18\f \x01(\v2\".spindle.agent.v1.CacheDrainResultR\x10cacheDrainResult\x126\n" + "\bpoweroff\x18\r \x01(\v2\x1a.spindle.agent.v1.PoweroffR\bpoweroff\x12I\n" + - "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xb9\x01\xbaH\xb5\x01\"\xb2\x01\n" + + "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult\x12J\n" + + "\x10open_debug_shell\x18\x0f \x01(\v2 .spindle.agent.v1.OpenDebugShellR\x0eopenDebugShell\x124\n" + + "\bpty_data\x18\x10 \x01(\v2\x19.spindle.agent.v1.PtyDataR\aptyData\x12:\n" + + "\n" + + "pty_resize\x18\x11 \x01(\v2\x1b.spindle.agent.v1.PtyResizeR\tptyResize:\xe1\x01\xbaH\xdd\x01\"\xda\x01\n" + "\x05hello\n" + "\x04init\n" + "\n" + @@ -1025,7 +1231,11 @@ "\vcache_drain\n" + "\x12cache_drain_result\n" + "\bpoweroff\n" + - "\x0fpoweroff_result\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" + "\x0fpoweroff_result\n" + + "\x10open_debug_shell\n" + + "\bpty_data\n" + + "\n" + + "pty_resize\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" var ( file_spindle_agent_v1_agent_proto_rawDescOnce sync.Once @@ -1039,7 +1249,7 @@ return file_spindle_agent_v1_agent_proto_rawDescData } -var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 14) +var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 17) var file_spindle_agent_v1_agent_proto_goTypes = []any{ (*Hello)(nil), // 0: spindle.agent.v1.Hello (*Init)(nil), // 1: spindle.agent.v1.Init @@ -1054,7 +1264,10 @@ (*CacheDrainResult)(nil), // 10: spindle.agent.v1.CacheDrainResult (*Poweroff)(nil), // 11: spindle.agent.v1.Poweroff (*PoweroffResult)(nil), // 12: spindle.agent.v1.PoweroffResult - (*Message)(nil), // 13: spindle.agent.v1.Message + (*OpenDebugShell)(nil), // 13: spindle.agent.v1.OpenDebugShell + (*PtyData)(nil), // 14: spindle.agent.v1.PtyData + (*PtyResize)(nil), // 15: spindle.agent.v1.PtyResize + (*Message)(nil), // 16: spindle.agent.v1.Message } var file_spindle_agent_v1_agent_proto_depIdxs = []int32{ 0, // 0: spindle.agent.v1.Message.hello:type_name -> spindle.agent.v1.Hello @@ -1070,11 +1283,14 @@ 10, // 10: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult 11, // 11: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff 12, // 12: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult - 13, // [13:13] is the sub-list for method output_type - 13, // [13:13] is the sub-list for method input_type - 13, // [13:13] is the sub-list for extension type_name - 13, // [13:13] is the sub-list for extension extendee - 0, // [0:13] is the sub-list for field type_name + 13, // 13: spindle.agent.v1.Message.open_debug_shell:type_name -> spindle.agent.v1.OpenDebugShell + 14, // 14: spindle.agent.v1.Message.pty_data:type_name -> spindle.agent.v1.PtyData + 15, // 15: spindle.agent.v1.Message.pty_resize:type_name -> spindle.agent.v1.PtyResize + 16, // [16:16] is the sub-list for method output_type + 16, // [16:16] is the sub-list for method input_type + 16, // [16:16] is the sub-list for extension type_name + 16, // [16:16] is the sub-list for extension extendee + 0, // [0:16] is the sub-list for field type_name } func init() { file_spindle_agent_v1_agent_proto_init() } @@ -1088,7 +1304,7 @@ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_agent_v1_agent_proto_rawDesc), len(file_spindle_agent_v1_agent_proto_rawDesc)), NumEnums: 0, - NumMessages: 14, + NumMessages: 17, NumExtensions: 0, NumServices: 0, }, diff --git a/spindle/engines/microvm/README.md b/spindle/engines/microvm/README.md --- a/spindle/engines/microvm/README.md +++ b/spindle/engines/microvm/README.md @@ -41,8 +41,13 @@ spindle expects, they should work. That is: - a guest agent is present inside of the image and when that image boots it will get started, -- `spindle-workflow` user exists, -- and the work directory is configured (`/workspace`). +- the `spindle-workflow` user exists, is unprivileged (non-zero uid/gid), and has + a usable login shell and home dir set in the image's passwd: workflow steps run + as this user, and the debug shell (see below) launches its passwd shell as a + login shell in its home dir. an unset or `nologin`/`false` shell breaks debug + ssh, +- and the work directory is configured (`/workspace`, with `/workspace/repo` as + the per-step working dir). ## Image discovery @@ -234,3 +239,27 @@ never made it to the destination store. The guest still only ever sees the same HTTP binary-cache upload protocol over vsock; it never gets direct access to SSH credentials or the destination store itself. + +### Debug ssh + +When a workflow fails, spindle can keep its microVM alive for a configured grace +window (`MicroVMPipelines.SSH`) and print an `ssh` invocation so you can poke at +the failed VM interactively. Spindle terminates the ssh connection itself and +bridges a pty into the live guest over the agent's vsock; the guest stays +keyless and never runs an ssh daemon. + +Access mirrors a git push: the ssh username is the job id, and the offered +public key is sent to the job's repo knot (`sh.tangled.repo.checkPushAllowed`). +The session is accepted only if that key is allowed to push to the job's repo. + +The shell is deliberately not configurable from either end. It always: +- runs as the `spindle-workflow` user (the ssh username selects the *job*, not a + unix user), +- uses that user's login shell from the image's passwd, launched as a login + shell (`-l`), and +- starts in the dir where the repo was cloned to. + +The only things the client influences are the terminal type and window size +(forwarded from the ssh pty request, and on resize). This relies on the image +configuring `spindle-workflow` properly per the expectations above; in +particular a missing or `nologin`/`false` won't work of course. diff --git a/spindle/engines/microvm/agent.go b/spindle/engines/microvm/agent.go --- a/spindle/engines/microvm/agent.go +++ b/spindle/engines/microvm/agent.go @@ -245,6 +245,19 @@ } } +func (s *AgentSession) OpenDebugShell(req *agentv1.OpenDebugShell) error { + s.mu.Lock() + defer s.mu.Unlock() + + if err := s.enc.Encode(&agentproto.Message{ + Id: "debug-shell", + OpenDebugShell: req, + }); err != nil { + return fmt.Errorf("send open_debug_shell: %w", err) + } + return nil +} + func (s *AgentSession) Poweroff(ctx context.Context) error { s.mu.Lock() defer s.mu.Unlock() diff --git a/spindle/engines/microvm/debug.go b/spindle/engines/microvm/debug.go new file mode 100644 --- /dev/null +++ b/spindle/engines/microvm/debug.go @@ -0,0 +1,326 @@ +package microvm + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "hash/fnv" + "io" + "log/slog" + "net" + "regexp" + "strings" + "sync" + "time" + + "tangled.org/core/spindle/agentproto" + agentv1 "tangled.org/core/spindle/agentproto/gen" + "tangled.org/core/spindle/models" +) + +const debugAcceptTimeout = 15 * time.Second + +type debugTarget struct { + cid uint32 + agent *AgentSession + knot string + repoDid string + wfLogger models.WorkflowLogger + maxAliveAt time.Time + stepCount int // index to emit the debug step at + connected chan struct{} // closed when the user first ssh's in, ending the grace window + released chan struct{} // closed when the user exits the debug shell, to tear down early +} + +var debugHandleSafe = regexp.MustCompile(`[^a-zA-Z0-9_.-]`) + +func newDebugHandle(wid models.WorkflowId) string { + h := fnv.New32a() + _, _ = io.WriteString(h, wid.String()) + token := hex.EncodeToString(h.Sum(nil)) + name := strings.Trim(debugHandleSafe.ReplaceAllString(wid.Name, "-"), "-") + if name == "" { + return token + } + return name + "-" + token +} + +func (e *Engine) registerDebugTarget(wid models.WorkflowId, t debugTarget) { + e.debugMu.Lock() + defer e.debugMu.Unlock() + e.debug[newDebugHandle(wid)] = t +} + +func (e *Engine) unregisterDebugTarget(wid models.WorkflowId) { + e.debugMu.Lock() + defer e.debugMu.Unlock() + delete(e.debug, newDebugHandle(wid)) +} + +// ends the grace window so we hold the VM until the shell exits instead. +func (e *Engine) markDebugConnected(handle string) { + e.debugMu.Lock() + defer e.debugMu.Unlock() + if t, ok := e.debug[handle]; ok && t.connected != nil { + select { + case <-t.connected: // already signalled + default: + close(t.connected) + } + } +} + +func (e *Engine) releaseDebugTarget(handle string) { + e.debugMu.Lock() + defer e.debugMu.Unlock() + t, ok := e.debug[handle] + if !ok { + return + } + delete(e.debug, handle) + if t.released != nil { + close(t.released) + } +} + +func (e *Engine) lookupDebugTarget(handle string) (debugTarget, bool) { + e.debugMu.Lock() + defer e.debugMu.Unlock() + t, ok := e.debug[handle] + return t, ok +} + +func (e *Engine) RepoForJob(jobID string) (knot, repoDid string, ok bool) { + t, found := e.lookupDebugTarget(jobID) + if !found || t.knot == "" || t.repoDid == "" { + return "", "", false + } + return t.knot, t.repoDid, true +} + +func (e *Engine) OpenDebugSession(ctx context.Context, jobID, term string, rows, cols int) (*DebugSession, error) { + target, ok := e.lookupDebugTarget(jobID) + if !ok { + return nil, fmt.Errorf("no live microVM for job %q", jobID) + } + + ln, port, err := listenRandomVsockPort(ctx) + if err != nil { + return nil, fmt.Errorf("listen for debug shell: %w", err) + } + filtered := &cidFilteredVsockListener{Listener: ln, cid: target.cid, logger: e.l} + + if err := target.agent.OpenDebugShell(&agentv1.OpenDebugShell{ + VsockPort: port, + Term: term, + Rows: clampDim(rows), + Cols: clampDim(cols), + }); err != nil { + _ = ln.Close() + return nil, fmt.Errorf("ask guest to open debug shell: %w", err) + } + + conn, err := acceptWithTimeout(ctx, filtered, debugAcceptTimeout) + if err != nil { + _ = ln.Close() + return nil, fmt.Errorf("accept debug shell connection: %w", err) + } + + e.markDebugConnected(jobID) + + return newDebugSession(conn, ln, e.l), nil +} + +func (e *Engine) maybeRetainForDebug(ctx context.Context, wid models.WorkflowId) { + handle := newDebugHandle(wid) + target, ok := e.lookupDebugTarget(handle) + if !ok { + // no target registered means the workflow didn't fail; nothing to retain + return + } + + wfLogger := target.wfLogger + if wfLogger == nil { + wfLogger = models.NullLogger{} + } + step := Step{name: "Debug shell", kind: models.StepKindSystem} + idx := target.stepCount + + wfLogger.ControlWriter(idx, step, models.StepStatusStart).Write([]byte{0}) + defer wfLogger.ControlWriter(idx, step, models.StepStatusEnd).Write([]byte{0}) + + ssh := e.cfg.MicroVMPipelines.DebugSSH + grace := ssh.GracePeriod + + cmd := debugSSHCommand(ssh.ListenAddr, e.cfg.Server.Hostname, handle) + out := wfLogger.DataWriter(idx, "stdout") + fmt.Fprintf(out, "Workflow failed, connect within %s to debug until shell exit or workflow timeout:\n", grace) + fmt.Fprintf(out, " %s\n", cmd) + e.l.Info("retaining failed microVM for debug", "workflow", wid, "grace", grace.String()) + + maxAlive := time.NewTimer(time.Until(target.maxAliveAt)) + defer maxAlive.Stop() + graceTimer := time.NewTimer(grace) + defer graceTimer.Stop() + + // wait for the user to ssh in within the grace window + select { + case <-ctx.Done(): + return + case <-maxAlive.C: + e.l.Info("debug retention hit max VM lifetime; tearing down microVM", "workflow", wid) + return + case <-graceTimer.C: + e.l.Info("nobody ssh'd in within grace; tearing down microVM", "workflow", wid) + return + case <-target.connected: + e.l.Info("debug shell connected; holding microVM until exit", "workflow", wid) + } + + // connected: hold the VM until the user exits or it hits its max lifetime + select { + case <-ctx.Done(): + case <-maxAlive.C: + e.l.Info("debug session hit max VM lifetime; tearing down microVM", "workflow", wid) + case <-target.released: + e.l.Info("debug shell exited; tearing down microVM", "workflow", wid) + } +} + +func debugSSHCommand(listenAddr, hostname, jobID string) string { + host, port := hostname, "" + if h, p, err := net.SplitHostPort(listenAddr); err == nil { + port = p + if h != "" && h != "0.0.0.0" && h != "::" { + host = h + } + } + cmd := "ssh -tt " + if port != "" && port != "22" { + cmd += "-p " + port + " " + } + return cmd + jobID + "@" + host +} + +// bridges an interactive shell over the agentproto vsock. +// Read to it gets the shell output from guest, Write sends the keyboard input from user. +type DebugSession struct { + conn net.Conn + ln net.Listener + enc *agentproto.Encoder + dec *agentproto.Decoder + l *slog.Logger + + out chan []byte + leftover []byte + exitCode int + closeOne sync.Once +} + +func newDebugSession(conn net.Conn, ln net.Listener, l *slog.Logger) *DebugSession { + d := &DebugSession{ + conn: conn, + ln: ln, + enc: agentproto.NewEncoder(conn), + dec: agentproto.NewDecoder(conn), + l: l, + out: make(chan []byte, 16), + } + go d.readLoop() + return d +} + +func (d *DebugSession) readLoop() { + defer close(d.out) + for { + msg, err := d.dec.Decode() + if err != nil { + if !errors.Is(err, io.EOF) { + d.l.Debug("debug shell decode ended", "error", err) + } + return + } + if p := msg.PtyData; p != nil && len(p.Data) > 0 { + d.out <- p.Data + } else if p := msg.ExecExit; p != nil { + d.exitCode = int(p.ExitCode) + return + } + } +} + +func (d *DebugSession) Read(p []byte) (int, error) { + if len(d.leftover) == 0 { + chunk, ok := <-d.out + if !ok { + return 0, io.EOF + } + d.leftover = chunk + } + n := copy(p, d.leftover) + d.leftover = d.leftover[n:] + return n, nil +} + +func (d *DebugSession) Write(p []byte) (int, error) { + if err := d.enc.Encode(&agentproto.Message{ + Id: "pty", + PtyData: &agentv1.PtyData{Data: append([]byte(nil), p...)}, + }); err != nil { + return 0, err + } + return len(p), nil +} + +func (d *DebugSession) Resize(rows, cols int) error { + return d.enc.Encode(&agentproto.Message{ + Id: "pty", + PtyResize: &agentv1.PtyResize{Rows: clampDim(rows), Cols: clampDim(cols)}, + }) +} + +func (d *DebugSession) ExitCode() int { return d.exitCode } + +func (d *DebugSession) Close() error { + var err error + d.closeOne.Do(func() { + err = d.conn.Close() + if d.ln != nil { + _ = d.ln.Close() + } + }) + return err +} + +func acceptWithTimeout(ctx context.Context, ln net.Listener, timeout time.Duration) (net.Conn, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + type result struct { + conn net.Conn + err error + } + ch := make(chan result, 1) + go func() { + conn, err := ln.Accept() + ch <- result{conn, err} + }() + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case r := <-ch: + return r.conn, r.err + } +} + +func clampDim(v int) uint32 { + if v < 1 { + return 1 + } + if v > 65535 { + return 65535 + } + return uint32(v) +} diff --git a/spindle/engines/microvm/debugssh.go b/spindle/engines/microvm/debugssh.go new file mode 100644 --- /dev/null +++ b/spindle/engines/microvm/debugssh.go @@ -0,0 +1,171 @@ +package microvm + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/pem" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "time" + + indigoxrpc "github.com/bluesky-social/indigo/xrpc" + "github.com/gliderlabs/ssh" + gossh "golang.org/x/crypto/ssh" + "tangled.org/core/api/tangled" + "tangled.org/core/hostutil" +) + +// debug ssh: terminates ssh here, bridges a pty into a live (failed) microVM +// over the guest's agent conn. the guest stays keyless. access mirrors git +// push: the offered key goes to the repo's knot +// (sh.tangled.repo.checkPushAllowed) and is accepted only if it can push to the +// job's repo. + +const debugAuthTimeout = 5 * time.Second + +func (e *Engine) serveDebugSSH(ctx context.Context) { + dbg := e.cfg.MicroVMPipelines.DebugSSH + if !dbg.Enabled || dbg.ListenAddr == "" { + return + } + addr := dbg.ListenAddr + httpc := &http.Client{Timeout: debugAuthTimeout} + + srv := &ssh.Server{ + Addr: addr, + Handler: e.debugHandle, + PublicKeyHandler: func(c ssh.Context, key ssh.PublicKey) bool { + return e.checkDebugAuth(c, c.User(), key, httpc) + }, + } + keyPath := e.cfg.MicroVMPipelines.DebugSSH.HostKeyPath + if keyPath == "" { + keyPath = filepath.Join(filepath.Dir(e.cfg.Server.DBPath), "debug_ssh_host_key") + } + if err := ensureDebugHostKey(keyPath); err != nil { + e.l.Error("debug ssh: ensure host key", "path", keyPath, "err", err) + return + } + if err := srv.SetOption(ssh.HostKeyFile(keyPath)); err != nil { + e.l.Error("debug ssh: load host key", "path", keyPath, "err", err) + return + } + + go func() { + <-ctx.Done() + _ = srv.Close() + }() + + e.l.Info("starting debug ssh server", "address", addr) + if err := srv.ListenAndServe(); err != nil && err != ssh.ErrServerClosed { + e.l.Error("debug ssh server stopped", "err", err) + } +} + +func ensureDebugHostKey(path string) error { + if _, err := os.Stat(path); err == nil { + return nil + } else if !os.IsNotExist(err) { + return fmt.Errorf("stat host key: %w", err) + } + + _, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + return fmt.Errorf("generate host key: %w", err) + } + block, err := gossh.MarshalPrivateKey(priv, "") + if err != nil { + return fmt.Errorf("marshal host key: %w", err) + } + + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("create host key dir: %w", err) + } + if err := os.WriteFile(path, pem.EncodeToMemory(block), 0o600); err != nil { + return fmt.Errorf("write host key: %w", err) + } + return nil +} + +func (e *Engine) debugHandle(sess ssh.Session) { + ptyReq, winCh, isPty := sess.Pty() + if !isPty { + io.WriteString(sess.Stderr(), "error: no terminal allocated; use `ssh -t`\n") + _ = sess.Exit(1) + return + } + + jobID := sess.User() + l := e.l.With("component", "debugssh", "job", jobID) + + debug, err := e.OpenDebugSession(sess.Context(), jobID, ptyReq.Term, ptyReq.Window.Height, ptyReq.Window.Width) + if err != nil { + fmt.Fprintf(sess.Stderr(), "error: %v\n", err) + _ = sess.Exit(1) + return + } + defer debug.Close() + l.Info("debug shell opened") + + go func() { + for win := range winCh { + if err := debug.Resize(win.Height, win.Width); err != nil { + l.Debug("debug ssh resize failed", "error", err) + } + } + }() + + // keyboard -> shell, runs until the client hangs up + go func() { _, _ = io.Copy(debug, sess) }() + // shell -> client, returns when the shell exits (Read hits EOF) + _, _ = io.Copy(sess, debug) + + code := debug.ExitCode() + l.Info("debug shell closed", "exitCode", code) + // the user is done; let retention tear the VM down now instead of waiting + // out the rest of the grace period + e.releaseDebugTarget(jobID) + _ = sess.Exit(code) +} + +func (e *Engine) checkDebugAuth(ctx context.Context, jobID string, key ssh.PublicKey, httpc *http.Client) bool { + l := e.l.With("component", "debugssh", "job", jobID, "keyType", key.Type()) + + knot, repoDid, ok := e.RepoForJob(jobID) + if !ok { + l.Warn("debug ssh: no live job / unknown repo") + return false + } + + host, noSSL, err := hostutil.ParseHostname(knot) + if err != nil { + l.Error("debug ssh: bad knot host", "knot", knot, "error", err) + return false + } + scheme := "https" + if noSSL { + scheme = "http" + } + xc := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, host), Client: httpc} + + reqCtx, cancel := context.WithTimeout(ctx, debugAuthTimeout) + defer cancel() + + out, err := tangled.RepoCheckPushAllowed(reqCtx, xc, string(gossh.MarshalAuthorizedKey(key)), repoDid) + if err != nil { + l.Error("debug ssh: push-allowed check failed", "knot", knot, "repo", repoDid, "error", err) + return false + } + if !out.Allowed { + l.Warn("debug ssh: key not allowed to push", "knot", knot, "repo", repoDid) + return false + } + if out.Did != nil { + l.Info("debug ssh: authorized", "did", *out.Did, "knot", knot, "repo", repoDid) + } + return true +} diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -53,6 +53,9 @@ cleanupMu sync.Mutex cleanup map[string][]cleanupFunc + + debugMu sync.Mutex + debug map[string]debugTarget } type Step struct { @@ -83,14 +86,21 @@ } } - return &Engine{ + e := &Engine{ l: l, cfg: cfg, db: d, scheduler: engine.NewResourceScheduler(budget, max, agingThreshold), cgroupParent: cgroupParent, cleanup: make(map[string][]cleanupFunc), - }, nil + debug: make(map[string]debugTarget), + } + + if cfg.MicroVMPipelines.DebugSSH.ListenAddr != "" { + go e.serveDebugSSH(ctx) + } + + return e, nil } func (e *Engine) ensureAgentHub() (*agentHub, error) { @@ -299,6 +309,7 @@ return err } state.VM = vm + state.StartedAt = time.Now() category = "Failed to connect to agent" @@ -323,6 +334,7 @@ return err } state.Agent = agentSession + state.CID = cid wf.Data = state e.registerCleanup(wid, func(ctx context.Context) error { @@ -397,6 +409,17 @@ if exitCode != 0 { e.l.Debug("step exited non-zero", "workflow", wid, "step", step.Name(), "exitCode", exitCode) + e.registerDebugTarget(wid, debugTarget{ + cid: state.CID, + agent: state.Agent, + knot: w.Environment["TANGLED_REPO_KNOT"], + repoDid: w.Environment["TANGLED_REPO_REPO_DID"], + wfLogger: wfLogger, + stepCount: len(w.Steps), + maxAliveAt: state.StartedAt.Add(e.WorkflowTimeout()), + connected: make(chan struct{}), + released: make(chan struct{}), + }) return fmt.Errorf("User step error: exited with code %d", exitCode) } return nil @@ -547,6 +570,11 @@ } func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { + if e.cfg.MicroVMPipelines.DebugSSH.Enabled { + // keep a failed VM alive for the grace period before we tear it down + e.maybeRetainForDebug(ctx, wid) + } + fns := e.drainCleanups(wid) var cleanupErr error @@ -557,10 +585,6 @@ } } return cleanupErr -} - -func (e *Engine) FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, w *models.Workflow, wfLogger models.WorkflowLogger) error { - return nil } func (e *Engine) WorkflowTimeout() time.Duration { diff --git a/spindle/engines/microvm/vm.go b/spindle/engines/microvm/vm.go --- a/spindle/engines/microvm/vm.go +++ b/spindle/engines/microvm/vm.go @@ -155,18 +155,23 @@ CacheReadURLs []string CacheTrustedPublicKeys []string VM VMHandle + CID uint32 Agent *AgentSession ReadCache *ReadCacheProxy UploadCache *UploadCacheProxy DNSProxy *DNSProxy WorkDir string NixOSToplevelCache nixosToplevelCacheStore + StartedAt time.Time // when the VM booted, for the max-lifetime cap } func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state *workflowState) error { if state == nil { return nil } + + // stop advertising this VM for debug shells before we tear it down + e.unregisterDebugTarget(wid) ctx = context.WithoutCancel(ctx) diff --git a/spindle/agentproto/spindle/agent/v1/agent.proto b/spindle/agentproto/spindle/agent/v1/agent.proto --- a/spindle/agentproto/spindle/agent/v1/agent.proto +++ b/spindle/agentproto/spindle/agent/v1/agent.proto @@ -82,13 +82,44 @@ string error = 1; } +message OpenDebugShell { + uint32 vsock_port = 1; + string term = 2; + uint32 rows = 3; + uint32 cols = 4; +} + +// changes meaning based on who sends this: +// guest->host is shell output, host->guest is keyboard input +message PtyData { + bytes data = 1; +} + +message PtyResize { + uint32 rows = 1; + uint32 cols = 2; +} + message Message { option (buf.validate.message).oneof = { fields: [ - "hello", "init", "exec_start", "exec_stdout", "exec_stderr", "exec_exit", - "activate_config", "activate_config_result", "built_paths", "cache_drain", - "cache_drain_result", "poweroff", "poweroff_result" - ], + "hello", + "init", + "exec_start", + "exec_stdout", + "exec_stderr", + "exec_exit", + "activate_config", + "activate_config_result", + "built_paths", + "cache_drain", + "cache_drain_result", + "poweroff", + "poweroff_result", + "open_debug_shell", + "pty_data", + "pty_resize" + ] required: true }; @@ -107,4 +138,7 @@ CacheDrainResult cache_drain_result = 12; Poweroff poweroff = 13; PoweroffResult poweroff_result = 14; + OpenDebugShell open_debug_shell = 15; + PtyData pty_data = 16; + PtyResize pty_resize = 17; } diff --git a/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs b/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs --- a/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs +++ b/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs @@ -113,6 +113,33 @@ pub error: ::prost::alloc::string::String, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct OpenDebugShell { + #[prost(uint32, tag = "1")] + pub vsock_port: u32, + #[prost(string, tag = "2")] + pub term: ::prost::alloc::string::String, + #[prost(uint32, tag = "3")] + pub rows: u32, + /// debug shells always run as the spindle-workflow user, with that user's + /// login shell, starting in its home dir. nothing here is client-specifiable. + #[prost(uint32, tag = "4")] + pub cols: u32, +} +/// changes meaning based on who sends this: +/// guest->host is shell output, host->guest is keyboard input +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct PtyData { + #[prost(bytes = "bytes", tag = "1")] + pub data: ::prost::bytes::Bytes, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct PtyResize { + #[prost(uint32, tag = "1")] + pub rows: u32, + #[prost(uint32, tag = "2")] + pub cols: u32, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct Message { #[prost(string, tag = "1")] pub id: ::prost::alloc::string::String, @@ -142,5 +169,11 @@ pub poweroff: ::core::option::Option, #[prost(message, optional, tag = "14")] pub poweroff_result: ::core::option::Option, + #[prost(message, optional, tag = "15")] + pub open_debug_shell: ::core::option::Option, + #[prost(message, optional, tag = "16")] + pub pty_data: ::core::option::Option, + #[prost(message, optional, tag = "17")] + pub pty_resize: ::core::option::Option, } // @@protoc_insertion_point(module) -- tangled.sh