diff --git a/client/src/commands.rs b/client/src/commands.rs index e826e9a..3b25ea2 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1088,6 +1088,16 @@ async fn job( let Some(permslip_url) = permslip_url else { return Err(CommandError::MissingPermslipUrl); }; + // Sign an interactive job for the sled its attachment + // will land on. + let target = if *interactive && target.single_baseboard().is_none() { + let Some(client) = client.as_ref() else { + return Err(CommandError::InteractiveTarget); + }; + Target::from(resolve_target(client, target).await?) + } else { + target.clone() + }; let mut signer = PermslipSigner::new(key_name, permslip_url).await?; let mut interval = interval(SIGNING_UPDATE_INTERVAL); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); @@ -1095,7 +1105,7 @@ async fn job( job_id.to_owned(), command, *interactive, - target.clone(), + target, )); pin!(sign); ctx.job_signing_started(&job_id); @@ -1842,6 +1852,8 @@ pub enum CommandError { IdentityMismatch { interactive: KeyId, key_id: KeyId }, #[error("❌ Interactive job error: {0}")] Interactive(#[from] InteractiveJobError), + #[error("❌ Interactive jobs must target exactly one sled")] + InteractiveTarget, #[error("❌ I/O error accessing `{path}`: {error}")] Io { path: PathBuf, diff --git a/common/src/jobs.rs b/common/src/jobs.rs index 2cf079c..f2fc2e6 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -28,7 +28,7 @@ use crate::borsh::{ }; use crate::interactive::InteractiveJobError; use crate::keys::{KeyId, Signed, ToBeSigned, Verified}; -use crate::targets::Target; +use crate::targets::{Cubbies, Target}; codephrase_newtype! { /// A globally unique identifier for a job within a session. @@ -211,6 +211,15 @@ impl JobStartRequest { } } + /// Interactive jobs run only on their single named baseboard. + pub fn runs_on(&self, baseboard: &BaseboardId, cubbies: &Cubbies) -> bool { + if self.interactive { + self.target.single_baseboard() == Some(baseboard) + } else { + self.target.includes(baseboard, cubbies) + } + } + pub fn job_id(&self) -> &JobId { &self.job_id } @@ -715,6 +724,29 @@ impl FromStr for JobOutputStream { #[cfg(test)] mod test { use super::*; + use crate::targets::SledId; + + /// Interactive jobs run only on their single named baseboard. + #[test] + fn interactive_targets() { + let sled = |serial: &str| BaseboardId { + part_number: "913".to_string(), + serial_number: serial.to_string(), + }; + let me = sled("me"); + let cubbies = Cubbies::from([(14, me.clone())]); + let job = |interactive, target| { + JobStartRequest::new(JobId::random(), "true", interactive, target) + }; + let just_me = Target::Sleds(vec![SledId::Baseboard(me.clone())]); + assert!(job(true, just_me.clone()).runs_on(&me, &cubbies)); + assert!(job(false, just_me).runs_on(&me, &cubbies)); + assert!(!job(true, Target::All).runs_on(&me, &cubbies)); + assert!(job(false, Target::All).runs_on(&me, &cubbies)); + let my_cubby = Target::Sleds(vec![SledId::Cubby(14)]); + assert!(!job(true, my_cubby.clone()).runs_on(&me, &cubbies)); + assert!(job(false, my_cubby).runs_on(&me, &cubbies)); + } /// Requested limits may narrow the default ceiling, never widen it. #[test] diff --git a/common/src/targets.rs b/common/src/targets.rs index b4cfde8..0a866f2 100644 --- a/common/src/targets.rs +++ b/common/src/targets.rs @@ -74,6 +74,23 @@ impl Target { pub fn is_all(&self) -> bool { matches!(self, Self::All) } + + /// A singular target or `None`. + pub fn single_baseboard(&self) -> Option<&BaseboardId> { + match self { + Self::All => None, + Self::Sleds(sleds) => match sleds.as_slice() { + [SledId::Baseboard(baseboard)] => Some(baseboard), + _ => None, + }, + } + } +} + +impl From for Target { + fn from(baseboard: BaseboardId) -> Self { + Self::Sleds(vec![SledId::Baseboard(baseboard)]) + } } impl FromStr for Target { diff --git a/server/src/error.rs b/server/src/error.rs index 619e93f..95455fb 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -38,6 +38,8 @@ pub enum JobError { InteractiveJob(#[from] InteractiveJobError), #[error("Command must not start with `-`")] InvalidCommand, + #[error("Interactive jobs must target exactly one sled")] + InteractiveTarget, #[error("Invalid range for output of length {0}")] InvalidRange(u64), #[error("I/O error during {what}: {error}")] @@ -176,7 +178,8 @@ impl From for HttpError { | SessionNotCurrent(_) => { HttpError::for_client_error(None, ClientErrorStatusCode::NOT_FOUND, message) } - DecodeCert(_) | DuplicateJobId(_) | InvalidCommand | Json(_) | MultipleSessions => { + DecodeCert(_) | DuplicateJobId(_) | InteractiveTarget | InvalidCommand | Json(_) + | MultipleSessions => { HttpError::for_client_error(None, ClientErrorStatusCode::BAD_REQUEST, message) } } diff --git a/server/src/manager.rs b/server/src/manager.rs index c60d1f1..d5f114e 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -597,6 +597,12 @@ impl JobManager { return Err(JobError::NoSession); } + // A broader target would orphan jobs on unattached sleds. + let payload = job.payload(); + if payload.interactive && payload.target().single_baseboard().is_none() { + return Err(JobError::InteractiveTarget); + } + // Reject job IDs we already know about, rather than silently // queuing a resubmission that can never advance the session's // job chain. Without this, a caller that resubmits an diff --git a/server/src/state.rs b/server/src/state.rs index e4a0299..202ab9c 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -203,7 +203,7 @@ impl<'a> SessionGuard<'a> { actor: &KeyId, ) { let job_id = *job.job_id(); - let targeted = job.payload().target().includes(own_baseboard, cubbies); + let targeted = job.payload().runs_on(own_baseboard, cubbies); if history.contains(&job_id) { // Note but otherwise ignore the duplicate job. info!(log, "already started job"; "job_id" => %job_id); @@ -277,7 +277,7 @@ impl<'a> SessionGuard<'a> { while let Some((request, params)) = self.next_queued_job() { let (tx_attachment, rx_attachment) = watch::channel(None); let job_id = request.payload().job_id().to_owned(); - if request.payload().target().includes(own_baseboard, cubbies) + if request.payload().runs_on(own_baseboard, cubbies) && history .get_job_status(&job_id) .map(|status| { diff --git a/tests/src/integration_tests.rs b/tests/src/integration_tests.rs index 247a21c..e7c6fdc 100644 --- a/tests/src/integration_tests.rs +++ b/tests/src/integration_tests.rs @@ -207,7 +207,7 @@ async fn client_proxy_server() { .expect("can't start session"); let job_id = session.next_job_id(); let job = root - .sign_job_request(&job_id, "cat > /dev/null", true) + .sign_job_request_for(&job_id, "cat > /dev/null", true, test_baseboard_id().into()) .await; let JobLimits { max_cpu, @@ -556,7 +556,7 @@ async fn interactive_job() { .expect("can't start session"); let job_id = session.next_job_id(); let job = root - .sign_job_request(&job_id, "cat > /dev/null", true) + .sign_job_request_for(&job_id, "cat > /dev/null", true, test_baseboard_id().into()) .await; let JobLimits { max_cpu, diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 6418129..c0c81d6 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -2110,3 +2110,23 @@ async fn job_signal_dispositions() { let stdout = String::from_utf8(stdout.to_vec()).unwrap(); assert!(stdout.contains("caught"), "stdout: {stdout:?}"); } + +/// Interactive jobs must target exactly one sled. +#[named] +#[tokio::test] +async fn interactive_target_required() { + let log = test_logger(function_name!()); + let (mgr, mut root, _dir, _shutdown) = manager_and_test_root(log).await; + let authn = fake_identity(&mut root).await; + let session_id = SessionId::random(); + let session = Session::new(session_id); + mgr.session_start(&authn, session_id, true).await.unwrap(); + + let job_id = session.next_job_id(); + let job = root.sign_job_request(&job_id, "bash", true).await; + assert!(matches!( + mgr.job_start(&authn, job.into_signed(), JobStartParams::default()) + .await, + Err(JobError::InteractiveTarget) + )); +}