From 63e3b5cb93e1399265605117853bf779f5c30bb1 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 3 Nov 2026 15:14:20 -0700 Subject: [PATCH] Check cubby targets before executing jobs Co-authored-by: Claude --- common/src/jobs.rs | 66 ++---------- common/src/targets.rs | 50 ++++----- server/src/manager.rs | 22 ++-- server/src/messages.rs | 20 +++- server/src/rack.rs | 83 ++++++++++++++- server/src/state.rs | 24 +++-- server/tests/distributed.rs | 98 +++++++++++------- .../tests/output/cubby-unknown-skip-event.bin | Bin 0 -> 147 bytes .../tests/output/not-targeted-skip-event.bin | Bin 0 -> 146 bytes sush.json | 4 +- 10 files changed, 217 insertions(+), 150 deletions(-) create mode 100644 server/tests/output/cubby-unknown-skip-event.bin create mode 100644 server/tests/output/not-targeted-skip-event.bin diff --git a/common/src/jobs.rs b/common/src/jobs.rs index c2eed34a..50de199b 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -4,7 +4,7 @@ //! Signed job requests. -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, HashMap}; use std::fmt; use std::io::Error as IoError; use std::ops::Deref; @@ -421,15 +421,6 @@ impl JobStartRequest { } } - /// Whether this job runs on `baseboard`. The signed target must match - /// `sleds`, so the sled that accepted the job cannot widen it. - /// Interactive jobs run only on their single named baseboard. - pub fn runs_on(&self, baseboard: &BaseboardId, sleds: &BTreeSet) -> bool { - sleds.contains(baseboard) - && self.target.matches(sleds) - && (!self.mode.is_interactive() || self.target.single_baseboard() == Some(baseboard)) - } - pub fn job_id(&self) -> &JobId { &self.job_id } @@ -449,6 +440,10 @@ impl JobStartRequest { pub fn target(&self) -> &Target { &self.target } + + pub fn target_matches_mode(&self) -> bool { + self.target.is_single() || !self.is_interactive() + } } impl ToBeSigned for JobStartRequest { @@ -524,6 +519,8 @@ pub enum JobStatus { pub enum SkipReason { SessionSuperseded, SessionEnded, + NotTargeted, + CubbyUnknown, } impl fmt::Display for SkipReason { @@ -531,6 +528,8 @@ impl fmt::Display for SkipReason { f.write_str(match self { Self::SessionSuperseded => "another session has superseded this one", Self::SessionEnded => "the session ended before the job could start", + Self::NotTargeted => "the job does not target this sled", + Self::CubbyUnknown => "this sled does not know its cubby yet", }) } } @@ -833,7 +832,6 @@ impl FromStr for JobOutputStream { mod test { use super::*; use crate::keys::{EccR, EccS, EncodedSignature}; - use crate::targets::SledId; /// A sequence number encodes to four words, parses back liberally, /// and orders numerically. @@ -1001,52 +999,6 @@ mod test { assert_eq!(server.next_job_id(), next); } - /// 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 sleds = BTreeSet::from([me.clone()]); - let job = |mode, target| { - JobStartRequest::new(JobId::random(), SessionId::random(), "true", mode, target) - }; - use JobMode::*; - let just_me = Target::Sleds(vec![SledId::Baseboard(me.clone())]); - assert!(job(Interactive, just_me.clone()).runs_on(&me, &sleds)); - assert!(job(Batch, just_me).runs_on(&me, &sleds)); - assert!(!job(Interactive, Target::All).runs_on(&me, &sleds)); - assert!(job(Batch, Target::All).runs_on(&me, &sleds)); - let my_cubby = Target::Sleds(vec![SledId::Cubby(14)]); - assert!(!job(Interactive, my_cubby.clone()).runs_on(&me, &sleds)); - assert!(job(Batch, my_cubby).runs_on(&me, &sleds)); - } - - /// A job runs only on the sleds its target resolved to, and only - /// if the signed target matches them. - #[test] - fn resolved_targets() { - let sled = |serial: &str| BaseboardId { - part_number: "913".to_string(), - serial_number: serial.to_string(), - }; - let (me, you) = (sled("me"), sled("you")); - let job = |target| { - JobStartRequest::new( - JobId::random(), - SessionId::random(), - "true", - JobMode::Batch, - target, - ) - }; - assert!(!job(Target::All).runs_on(&me, &BTreeSet::from([you.clone()]))); - let just_you = Target::from(you.clone()); - assert!(!job(just_you).runs_on(&me, &BTreeSet::from([me.clone(), you]))); - } - /// Requested limits may narrow the default ceiling, never widen it. #[test] fn limits_clamp() { diff --git a/common/src/targets.rs b/common/src/targets.rs index 5d582a05..c5c73f43 100644 --- a/common/src/targets.rs +++ b/common/src/targets.rs @@ -7,8 +7,8 @@ //! One grammar serves signed job requests, the client, and the proxy: //! `*` means every sled in the rack, and a comma-separated list names //! sleds by baseboard ID or cubby number. The sled that accepts a job -//! resolves its target once, to the baseboards it can reach, and every -//! sled runs the job against that resolution. +//! resolves its target once, to the baseboards it can reach. Sleds that +//! receive the job only run it if the signed target includes them. use std::collections::{BTreeMap, BTreeSet}; use std::fmt; @@ -104,22 +104,14 @@ impl Target { } } - /// Whether every sled in `sleds` is one this target names. Each cubby - /// matches any one sled. - pub fn matches(&self, sleds: &BTreeSet) -> bool { + /// Whether this target includes `baseboard` or `cubby`. + pub fn matches(&self, baseboard: &BaseboardId, cubby: Option) -> bool { match self { Self::All => true, - Self::Sleds(named) => { - let mut baseboards = BTreeSet::new(); - let mut cubbies = BTreeSet::new(); - for sled in named { - match sled { - SledId::Baseboard(baseboard) => baseboards.insert(baseboard.clone()), - SledId::Cubby(cubby) => cubbies.insert(*cubby), - }; - } - sleds.difference(&baseboards).count() <= cubbies.len() - } + Self::Sleds(sleds) => sleds.iter().any(|sled| match sled { + SledId::Baseboard(listed) => listed == baseboard, + SledId::Cubby(listed) => Some(*listed) == cubby, + }), } } @@ -311,21 +303,15 @@ mod test { #[test] fn targets_match() { let brm31 = baseboard("913-0000019:BRM42220031"); - let brm40 = baseboard("913-0000019:BRM42220040"); - let both = BTreeSet::from([brm31.clone(), brm40.clone()]); - let matches = |target: &str, sleds: &BTreeSet| { - target.parse::().unwrap().matches(sleds) - }; - - assert!(matches("*", &both)); - assert!(matches( - "913-0000019:BRM42220031", - &BTreeSet::from([brm31.clone()]) - )); - assert!(matches("913-0000019:BRM42220031", &BTreeSet::new())); - assert!(!matches("913-0000019:BRM42220031", &both)); - assert!(matches("14", &BTreeSet::from([brm40.clone()]))); - assert!(!matches("14", &both)); - assert!(matches("913-0000019:BRM42220031,14", &both)); + let matches = + |target: &str, cubby| target.parse::().unwrap().matches(&brm31, cubby); + + assert!(matches("*", None)); + assert!(matches("913-0000019:BRM42220031", None)); + assert!(!matches("913-0000019:BRM42220040", Some(14))); + assert!(matches("913-0000019:BRM42220040,14", Some(14))); + assert!(matches("14", Some(14))); + assert!(!matches("14", Some(15))); + assert!(!matches("14", None)); } } diff --git a/server/src/manager.rs b/server/src/manager.rs index 56801f36..b62c3d0c 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -420,30 +420,26 @@ impl JobManager { job: SignedJob, params: JobStartParams, ) -> Result, JobError> { - // A forged job must not take a signed job's place in the chain. + // Check the job signature. self.state .borrow() .certs() .verify(&job) .map_err(JobError::UnverifiedJob)?; - self.session_check(job.payload().session_id())?; - // A broader target would orphan jobs on unattached sleds. - let payload = job.payload(); - if payload.is_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 - // already-completed job ID and waits would see the *original* - // job's status returned as if it belonged to this submission. + // Reject jobs we already know about. let job_id = job.job_id().to_owned(); if self.state.borrow().get_job_status(&job_id).is_some() { return Err(JobError::DuplicateJobId(job_id)); } + // Check the session and target. + let payload = job.payload(); + self.session_check(payload.session_id())?; + if !payload.target_matches_mode() { + return Err(JobError::InteractiveTarget); + } + // Submit the job for execution. let sleds = self.reachable_targets(payload.target())?; let wait = params.wait.to_owned(); diff --git a/server/src/messages.rs b/server/src/messages.rs index 5e2efd6c..7b3a4e48 100644 --- a/server/src/messages.rs +++ b/server/src/messages.rs @@ -331,6 +331,8 @@ mod wire_format { Event::Job(JobEvent::Skipped(_, _, reason)) => match reason { SkipReason::SessionSuperseded => "job-skipped-event", SkipReason::SessionEnded => "session-ended-skip-event", + SkipReason::NotTargeted => "not-targeted-skip-event", + SkipReason::CubbyUnknown => "cubby-unknown-skip-event", }, }, } @@ -477,13 +479,29 @@ mod wire_format { )), ), Message::Event( - baseboard, + baseboard.clone(), Event::Job(JobEvent::Skipped( skip_job, skip_when, SkipReason::SessionEnded, )), ), + Message::Event( + baseboard.clone(), + Event::Job(JobEvent::Skipped( + skip_job, + skip_when, + SkipReason::NotTargeted, + )), + ), + Message::Event( + baseboard, + Event::Job(JobEvent::Skipped( + skip_job, + skip_when, + SkipReason::CubbyUnknown, + )), + ), ] } diff --git a/server/src/rack.rs b/server/src/rack.rs index 9ed163b6..5d1a3b2e 100644 --- a/server/src/rack.rs +++ b/server/src/rack.rs @@ -8,7 +8,7 @@ use std::collections::BTreeSet; use sled_hardware_types::BaseboardId; -use sush_common::jobs::SignedJob; +use sush_common::jobs::{JobStartRequest, SkipReason}; use sush_common::targets::Cubbies; use sush_common::version::{VersionInfo, VersionMap}; @@ -17,6 +17,11 @@ pub struct Rack { /// The ID of the baseboard (sled) this server is running on. own_baseboard: BaseboardId, + /// The cubby this sled occupies, once MGS has reported it. + /// Does not change once known, since a sled cannot change + /// cubbies without rebooting. + own_cubby: Option, + /// Sled baseboards by cubby number, as reported by MGS. cubbies: Cubbies, @@ -32,13 +37,24 @@ impl Rack { Self { versions: [(own_baseboard.clone(), VersionInfo::current())].into(), own_baseboard, + own_cubby: None, cubbies: Default::default(), unknown_versions: Default::default(), } } - pub fn runs_here(&self, job: &SignedJob, sleds: &BTreeSet) -> bool { - job.runs_on(&self.own_baseboard, sleds) + /// Why `job` should not run on this sled, or `None` if it should. + /// The sled that accepted the job resolved its target to this sled, + /// but that resolution is unsigned so we must check it ourselves. + pub fn check_target(&self, job: &JobStartRequest) -> Option { + let target = job.target(); + if job.target_matches_mode() && target.matches(&self.own_baseboard, self.own_cubby) { + None + } else if self.own_cubby.is_none() { + Some(SkipReason::CubbyUnknown) + } else { + Some(SkipReason::NotTargeted) + } } pub fn own_baseboard(&self) -> &BaseboardId { @@ -50,6 +66,12 @@ impl Rack { } pub fn set_cubbies(&mut self, cubbies: Cubbies) { + if let Some((&cubby, _)) = cubbies + .iter() + .find(|(_, baseboard)| **baseboard == self.own_baseboard) + { + self.own_cubby = Some(cubby); + } self.cubbies = cubbies; } @@ -72,3 +94,58 @@ impl Rack { self.unknown_versions.insert(version) } } + +#[cfg(test)] +mod test { + use sush_common::jobs::{JobId, JobMode, SessionId}; + use sush_common::targets::Target; + + use super::*; + + #[test] + fn check_target() { + let me: BaseboardId = "913-0000019:BRM42220031".parse().unwrap(); + let mut rack = Rack::new(me.clone()); + let check_target = |rack: &Rack, mode, target: &str| { + rack.check_target(&JobStartRequest::new( + JobId::random(), + SessionId::random(), + "true", + mode, + target.parse::().unwrap(), + )) + }; + + use JobMode::*; + use SkipReason::*; + + // Before this sled knows its cubby. + assert_eq!(check_target(&rack, Batch, "*"), None); + assert_eq!(check_target(&rack, Batch, "913-0000019:BRM42220031"), None); + assert_eq!(check_target(&rack, Batch, "14"), Some(CubbyUnknown)); + + // Once a cubby map places this sled in cubby 14, the sled keeps + // that cubby, even after a later map that omits it. + rack.set_cubbies(Cubbies::from([(14, me.clone())])); + rack.set_cubbies(Cubbies::new()); + assert_eq!(check_target(&rack, Batch, "14"), None); + assert_eq!(check_target(&rack, Batch, "15"), Some(NotTargeted)); + assert_eq!( + check_target(&rack, Batch, "913-0000019:BRM42220040"), + Some(NotTargeted) + ); + + // Interactive jobs run only on the single sled they target. + assert_eq!( + check_target(&rack, Interactive, "913-0000019:BRM42220031"), + None + ); + assert_eq!(check_target(&rack, Interactive, "14"), None); + assert_eq!(check_target(&rack, Interactive, "15"), Some(NotTargeted)); + assert_eq!(check_target(&rack, Interactive, "*"), Some(NotTargeted)); + assert_eq!( + check_target(&rack, Interactive, "913-0000019:BRM42220031,14"), + Some(NotTargeted) + ); + } +} diff --git a/server/src/state.rs b/server/src/state.rs index eac417bf..eb0b57b0 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -396,7 +396,7 @@ impl State { fn skip_queued_jobs(&self, now: DateTime) -> Vec { let mut outputs = Vec::new(); for (job_id, queued) in self.session.queued_jobs().into_iter().flatten() { - if !queued.replayed && self.rack.runs_here(&queued.job, &queued.sleds) { + if !queued.replayed && queued.sleds.contains(self.rack.own_baseboard()) { outputs.push(Output::job_event( self.rack.own_baseboard(), JobEvent::Skipped(*job_id, now, SkipReason::SessionEnded), @@ -836,7 +836,9 @@ impl State { "session_id" => %session_id, "actor" => %actor, ); - if live && self.rack.runs_here(signed, sleds) && !self.history.contains(&job_id) + if live + && sleds.contains(self.rack.own_baseboard()) + && !self.history.contains(&job_id) { outputs.push(Output::job_event( self.rack.own_baseboard(), @@ -1032,7 +1034,7 @@ impl State { }; let job_id = *queued.job.job_id(); let replayed = queued.replayed; - let targeted = self.rack.runs_here(&queued.job, &queued.sleds); + let targeted = queued.sleds.contains(self.rack.own_baseboard()); // Adjudication can give the boundary job a status before its // request replays. The queue releases jobs in order, so @@ -1075,7 +1077,7 @@ impl State { return; }; let cancel_here = match queued_jobs.remove(job_id) { - Some(queued) => self.rack.runs_here(&queued.job, &queued.sleds), + Some(queued) => queued.sleds.contains(self.rack.own_baseboard()), None => !self.history.contains(job_id), }; if !replayed && cancel_here { @@ -1147,10 +1149,18 @@ impl State { replayed, } = queued_jobs.remove(&next_id).expect("checked above"); let job_id = request.payload().job_id().to_owned(); - if self.rack.runs_here(&request, &sleds) { - if matches!(admission, Admission::Refuse) { + if sleds.contains(self.rack.own_baseboard()) { + if let Some(reason) = self.rack.check_target(request.payload()) { if !replayed { - warn!(log, "skipping a superseded session's job"; "job_id" => %job_id); + warn!(log, "refusing to execute job"; "job_id" => %job_id, "reason" => %reason); + outputs.push(Output::job_event( + self.rack.own_baseboard(), + JobEvent::Skipped(job_id, now, reason), + )); + } + } else if matches!(admission, Admission::Refuse) { + if !replayed { + warn!(log, "skipping job from superseded session"; "job_id" => %job_id); outputs.push(Output::job_event( self.rack.own_baseboard(), JobEvent::Skipped(job_id, now, SkipReason::SessionSuperseded), diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 1fce6176..a746f059 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -298,12 +298,12 @@ async fn first_job_confirms_start_on_another_sled() { shutdown.cancel(); } -/// A job that names a cubby runs on the sled in that cubby even while -/// that sled's own cubby map is empty, because the sled that accepts -/// the job resolves the cubby. +/// The sled that accepts a cubby-targeted job resolves the cubby, but +/// the listed sled runs the job only if it knows its own cubby and the +/// signed target names it. Otherwise the listed sled skips the job. #[named] #[tokio::test] -async fn cubby_resolves_on_accepting_sled() { +async fn listed_sled_checks_own_cubby() { let (_tmp, dir) = pki("sush-distributed-", 2); let mut root = common::ephemeral_root(); let root_pem = dir.join("job-root.pem"); @@ -325,11 +325,6 @@ async fn cubby_resolves_on_accepting_sled() { .await; sees(&a, &b).await; - // Only A knows that B is in cubby 14. - a.cubbies - .send(Cubbies::from([(14, b.baseboard.clone())])) - .unwrap(); - let authn = requester(&root); let Ceremony { session_id, @@ -341,38 +336,69 @@ async fn cubby_resolves_on_accepting_sled() { .session_start(&authn, session_id, client_nonce, signer_nonce, seq) .await .unwrap(); - let job_id = Session::new(session_id).next_job_id(); - let job = root - .sign(JobStartRequest::new( - job_id, - session_id, - "true", - JobMode::Batch, - "14".parse().unwrap(), - )) - .await - .unwrap(); + let mut session = Session::new(session_id); + // Start a job naming `cubby` on A, and return B's status for it. // A's map update lands whenever its state task gets to it, so // retry the job until A accepts it. - let wait = JobStartParams { - wait: JobWait::Stop, - ..Default::default() + let mut run = async |cubby: u8| { + let job = root + .sign(JobStartRequest::new( + session.next_job_id(), + session_id, + "true", + JobMode::Batch, + cubby.to_string().parse().unwrap(), + )) + .await + .unwrap(); + let wait = JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }; + let sleds = loop { + match a.mgr.job_start(&authn, job.clone(), wait.clone()).await { + Ok(sleds) => break sleds, + Err(JobError::UnknownCubby(_)) => sleep(Duration::from_millis(50)).await, + Err(error) => panic!("job start failed: {error}"), + } + }; + assert_eq!(sleds, BTreeSet::from([b.baseboard.clone()])); + let mut status = a.mgr.job_status(&authn, job.job_id()).await.unwrap(); + assert!(!status.contains_key(&a.baseboard)); + session.job_started(job); + status.remove(&b.baseboard).unwrap() }; - let sleds = loop { - match a.mgr.job_start(&authn, job.clone(), wait.clone()).await { - Ok(sleds) => break sleds, - Err(JobError::UnknownCubby(14)) => sleep(Duration::from_millis(50)).await, - Err(error) => panic!("job start failed: {error}"), - } + let skipped = |status: &JobStatus| match status { + JobStatus::Skipped { reason, .. } => Some(*reason), + _ => None, }; - assert_eq!(sleds, BTreeSet::from([b.baseboard.clone()])); - let status = a.mgr.job_status(&authn, &job_id).await.unwrap(); - assert!(matches!( - status.get(&b.baseboard), - Some(JobStatus::Stopped { .. }) - )); - assert!(!status.contains_key(&a.baseboard)); + + // Only A knows that B is in cubby 14, so B cannot tell whether the + // job names it. + a.cubbies + .send(Cubbies::from([(14, b.baseboard.clone())])) + .unwrap(); + assert_eq!(skipped(&run(14).await), Some(SkipReason::CubbyUnknown)); + + // B knows it is in cubby 15, so A's map is wrong. + b.cubbies + .send(Cubbies::from([(15, b.baseboard.clone())])) + .unwrap(); + eventually("B learns its cubby", 60, async || { + b.mgr + .versions() + .iter() + .any(|row| row.baseboard == b.baseboard && row.cubby == Some(15)) + }) + .await; + assert_eq!(skipped(&run(14).await), Some(SkipReason::NotTargeted)); + + // With A's map corrected, the job names B and runs there. + a.cubbies + .send(Cubbies::from([(15, b.baseboard.clone())])) + .unwrap(); + assert!(matches!(run(15).await, JobStatus::Stopped { .. })); shutdown.cancel(); } diff --git a/server/tests/output/cubby-unknown-skip-event.bin b/server/tests/output/cubby-unknown-skip-event.bin new file mode 100644 index 0000000000000000000000000000000000000000..e76a57da30ef6d5307ce165df05e7490bce21ba0 GIT binary patch literal 147 zcmZ3O6lSn6)wL`&ucT>Fc0poMNqk;uZc=Jdwxyx5t^o)bTILp~7G);pz+{|)d`*mu zj0_Bn4HhPQFc0poMNqk;uZc=Jdwxyx5t^o)bTILp~7G);pz+{|)d`*mu zj0_Bn4HhPQ