Skip to content
2 changes: 1 addition & 1 deletion api/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -388,7 +388,7 @@ impl JobWait {
match self {
Self::None => true,
Self::Start => !matches!(status, Queued { .. }),
Self::Stop => matches!(status, Cancelled { .. } | Error { .. } | Stopped { .. }),
Self::Stop => status.is_terminal(),
}
}
}
Expand Down
32 changes: 25 additions & 7 deletions client/src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ impl Cli {
let path = match BaseDirectories::with_prefix(PREFIX).place_state_file(SESSION_FILE_NAME) {
Ok(path) => path,
Err(error) => {
eprintln!("⚠️ The session will not persist: {error}");
eprintln!(" The session will not persist: {error}");
return;
}
};
Expand All @@ -102,17 +102,17 @@ impl Cli {
session,
}) => *self.session.lock().unwrap() = Some(session),
Ok(SavedSession { version, .. }) => {
eprintln!("⚠️ Ignoring a version {version} saved session")
eprintln!(" Ignoring a version {version} saved session")
}
Err(error) => eprintln!("⚠️ Ignoring the saved session: {error}"),
Err(error) => eprintln!(" Ignoring the saved session: {error}"),
},
Err(error) if error.kind() == ErrorKind::NotFound => (),
Err(error) => eprintln!("⚠️ Ignoring the saved session: {error}"),
Err(error) => eprintln!(" Ignoring the saved session: {error}"),
}
self.session_file = Some(path);
match BaseDirectories::with_prefix(PREFIX).place_state_file(TOKEN_FILE_NAME) {
Ok(path) => self.token_file = Some(path),
Err(error) => eprintln!("⚠️ Signing tokens will not persist: {error}"),
Err(error) => eprintln!(" Signing tokens will not persist: {error}"),
}
}

Expand Down Expand Up @@ -149,7 +149,7 @@ impl Cli {
},
};
if let Err(error) = result {
eprintln!("⚠️ The session was not saved: {error}");
eprintln!(" The session was not saved: {error}");
}
}
}
Expand Down Expand Up @@ -203,6 +203,11 @@ fn short_status_row(status: &JobStatus) -> String {
JobStatus::Error {
time_error, error, ..
} => format!("Error at {time_error}: {error}"),
JobStatus::Skipped {
time_skipped,
reason,
..
} => format!("Skipped at {time_skipped}: {reason}"),
}
}

Expand Down Expand Up @@ -304,7 +309,7 @@ impl CommandContext for Cli {
.map_err(io::Error::other)
.and_then(|json| write_private(path, &json));
if let Err(error) = result {
eprintln!("⚠️ The token was not saved: {error}");
eprintln!(" The token was not saved: {error}");
}
}

Expand Down Expand Up @@ -900,6 +905,19 @@ impl CommandContext for Cli {
Error:\t{error}"
)
}
JobStatus::Skipped {
job_id,
time_skipped,
reason,
} => {
println!(
"⏩ Job ID:\t{job_id}\n \
Target:\t{baseboard_id}\n \
Job status:\tSkipped\n \
Skipped at:\t{time_skipped}\n \
Reason:\t{reason}"
)
}
}
}
}
Expand Down
7 changes: 6 additions & 1 deletion client/src/commands.rs
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ use sush_common::interactive::{InteractiveJobError, InteractiveJobMessage};
use sush_common::jobs::JobOutputStream::{self, Stderr, Stdout};
use sush_common::jobs::{
Access, JobId, JobLimits, JobMode, JobOutputHash, JobOutputState, JobStatus, JobStatusMap,
Session, SessionId, SessionSignerNonce, SignedJob, job_status_try_from_json_map,
Session, SessionId, SessionSignerNonce, SignedJob, SkipReason, job_status_try_from_json_map,
};
#[cfg(feature = "permslip")]
use sush_common::jobs::{JobStartRequest, SessionSushNonce};
Expand Down Expand Up @@ -2073,6 +2073,9 @@ async fn job_output_from(
Some(JobStatus::Started { job_id, .. }) => {
return Err(CommandError::JobStillRunning(job_id.to_owned()));
}
Some(JobStatus::Skipped { job_id, reason, .. }) => {
return Err(CommandError::JobSkipped(job_id.to_owned(), *reason));
}
Some(JobStatus::Stopped { output, .. }) => output,
};
let len = match stream {
Expand Down Expand Up @@ -2526,6 +2529,8 @@ pub enum CommandError {
JobDidNotRun(JobId),
#[error("❌ Job `{0}` is not yet running")]
JobNotYetRunning(JobId),
#[error("⏩ Job `{0}` was skipped on this sled: {1}")]
JobSkipped(JobId, SkipReason),
#[error("❌ Job `{0}` is still running")]
JobStillRunning(JobId),
#[error("❌ JSON error: {0}")]
Expand Down
2 changes: 1 addition & 1 deletion client/src/tunnel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ impl Tunnel {
let target = Arc::clone(&target);
connections.spawn(async move {
if let Err(error) = forward(&target, stream).await {
eprintln!("⚠️ Tunnel connection failed: {error}");
eprintln!(" Tunnel connection failed: {error}");
}
});
}
Expand Down
22 changes: 20 additions & 2 deletions common/src/authn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ impl RequestKey {

codephrase_newtype! {
/// The server half of an ephemeral request-signing key.
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[derive(Clone, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
pub struct RequestVerifier = Full;

}
Expand Down Expand Up @@ -347,7 +347,7 @@ impl SeqWindow {
/// Response to an authentication challenge, containing the server-chosen
/// nonce and a fresh client-chosen nonce. This is the structure that is
/// signed and verified as authentication credentials.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
pub struct ChallengeResponse {
nonce: Nonce,
cnonce: Nonce,
Expand Down Expand Up @@ -613,6 +613,24 @@ mod test {

use super::*;

/// A login is signed, gossiped, and re-verified by every sled,
/// which rebuilds it from the wire to check the signature. Two
/// shapes in one rack disagree about which logins verify. This
/// pin freezes the shape: any field change fails here. Do not
/// re-pin; add a new wire version.
#[test]
fn pin_challenge_response_schema() {
let schema =
serde_json::to_string_pretty(&schemars::schema_for!(ChallengeResponse)).unwrap();
let path = "tests/output/challenge-response-schema.json";
if std::env::var("EXPECTORATE").as_deref() == Ok("overwrite") {
std::fs::write(path, &schema).unwrap();
} else {
let expected = std::fs::read_to_string(path).expect("missing snapshot");
assert_eq!(schema, expected, "the signed login's shape changed");
}
}

/// Values to be signed must match even across versions.
#[test]
fn pin_to_be_signed() {
Expand Down
81 changes: 73 additions & 8 deletions common/src/jobs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ impl SessionId {
LastJob::None => hash(&[b"None", self.0.to_be_bytes().as_slice()].concat()),
LastJob::Some(job) => hash(&[b"Some", job.to_be_signed().as_slice()].concat()),
LastJob::Burned(job_id) => hash(&[b"Burned", job_id.to_be_bytes().as_slice()].concat()),
LastJob::Resumed(next) => return *next,
})
}
}
Expand Down Expand Up @@ -153,6 +154,7 @@ pub enum LastJob {
None,
Some(SignedJob),
Burned(JobId),
Resumed(JobId),
}

#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
Expand Down Expand Up @@ -201,11 +203,13 @@ impl Session {
self.last_job = LastJob::Some(job)
}

/// Burn `job_id`, either as the session's next job or by
/// rewinding it from the chain head. The rewind unwinds a
/// signed-but-unrun job on a signer. On a server it converges a
/// skip that raced the start it names, keeping the execution in
/// history. Returns whether the chain moved.
/// Burn `job_id` when it is the session's next job or the job at
/// the chain head. Burning the next job skips it before it runs.
/// Burning the head rewrites the chain to continue from the burn:
/// a signer unwinds a job it signed that never ran, and a server
/// converges with that signer when a skip request arrives after
/// the start of the job it names. An execution already in history
/// stays there. Returns whether the chain moved.
pub fn skip_job(&mut self, job_id: JobId) -> bool {
if job_id == self.next_job_id()
|| matches!(&self.last_job, LastJob::Some(job) if *job.job_id() == job_id)
Expand All @@ -220,6 +224,14 @@ impl Session {
pub fn next_job_id(&self) -> JobId {
self.session_id.next_job_id(&self.last_job)
}

/// Resume the chain at `successor`, the position a boundary
/// record stored at its last commitment. Every position before
/// `successor` was already handled by the sled that stored the
/// record.
pub fn resume_at(&mut self, successor: JobId) {
self.last_job = LastJob::Resumed(successor);
}
}

/// How a job runs. The streaming modes allow **unrecorded** I/O.
Expand Down Expand Up @@ -375,6 +387,37 @@ pub enum JobStatus {
result: Result<i32, ProcessError>,
output: JobOutputState,
},
/// The reporting sled decided it will never run this job. A skip
/// is a decision, not a failure: the job may have run on other
/// sleds, and the operator decides whether to resubmit.
Skipped {
job_id: JobId,
time_skipped: DateTime<Utc>,
reason: SkipReason,
},
}

/// Why a sled will never run a job.
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum SkipReason {
/// The job's session sits at or below the sled's execution floor,
/// where the sled cannot tell replay from re-run.
BelowFloor,
/// The job's chain position precedes the sled's recorded
/// commitment: a previous life already handled it.
AlreadyHandled,
SessionEnded,
}

impl fmt::Display for SkipReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::BelowFloor => "the session sits below this sled's execution floor",
Self::AlreadyHandled => "a previous life of this sled already handled it",
Self::SessionEnded => "the session ended before the job could start",
})
}
}

pub type JobStatusMap = BTreeMap<BaseboardId, JobStatus>;
Expand Down Expand Up @@ -476,7 +519,10 @@ impl JobStatus {
pub fn is_terminal(&self) -> bool {
matches!(
self,
Self::Cancelled { .. } | Self::Error { .. } | Self::Stopped { .. }
Self::Cancelled { .. }
| Self::Error { .. }
| Self::Stopped { .. }
| Self::Skipped { .. }
)
}

Expand All @@ -488,12 +534,13 @@ impl JobStatus {
Self::Error { time_error, .. } => *time_error,
Self::Started { time_started, .. } => *time_started,
Self::Stopped { time_stopped, .. } => *time_stopped,
Self::Skipped { time_skipped, .. } => *time_skipped,
}
}

pub fn time_elapsed(&self) -> TimeDelta {
match self {
Self::Cancelled { .. } | Self::Error { .. } => TimeDelta::zero(),
Self::Cancelled { .. } | Self::Error { .. } | Self::Skipped { .. } => TimeDelta::zero(),
Self::Queued { time_queued, .. } => Utc::now() - time_queued,
Self::Started { time_started, .. } => Utc::now() - time_started,
Self::Stopped {
Expand All @@ -510,7 +557,8 @@ impl JobStatus {
| Self::Queued { job_id, .. }
| Self::Error { job_id, .. }
| Self::Started { job_id, .. }
| Self::Stopped { job_id, .. } => job_id,
| Self::Stopped { job_id, .. }
| Self::Skipped { job_id, .. } => job_id,
}
}

Expand Down Expand Up @@ -685,6 +733,23 @@ mod test {
use crate::keys::{EccR, EccS, EncodedSignature};
use crate::targets::SledId;

/// A job request is signed, and every sled rebuilds it from the
/// wire to check the signature, so two sleds with different
/// request shapes disagree about what verifies. This pin freezes
/// the shape: any field change fails here. Do not re-pin; add a
/// new wire version.
#[test]
fn pin_job_start_request_schema() {
let schema = serde_json::to_string_pretty(&schemars::schema_for!(JobStartRequest)).unwrap();
let path = "tests/output/job-start-request-schema.json";
if std::env::var("EXPECTORATE").as_deref() == Ok("overwrite") {
std::fs::write(path, &schema).unwrap();
} else {
let expected = std::fs::read_to_string(path).expect("missing snapshot");
assert_eq!(schema, expected, "the signed request's shape changed");
}
}

/// A request's defaulted fields stay out of the signed material,
/// so a signature made before a field existed still verifies
/// after it is added. The literal hash pins the scheme for
Expand Down
8 changes: 8 additions & 0 deletions common/src/keys.rs
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,14 @@ impl<'de> Deserialize<'de> for SshPublicKey {
}

impl SshPublicKey {
pub fn to_openssh(&self) -> Result<String, KeyError> {
Ok(self.0.to_openssh()?)
}

pub fn from_openssh(openssh: &str) -> Result<Self, KeyError> {
Ok(Self(ssh_key::PublicKey::from_openssh(openssh)?))
}

pub fn key_id(&self) -> Result<KeyId, KeyError> {
KeyId::try_from(&self.0)
}
Expand Down
32 changes: 32 additions & 0 deletions common/tests/output/challenge-response-schema.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "ChallengeResponse",
"description": "Response to an authentication challenge, containing the server-chosen nonce and a fresh client-chosen nonce. This is the structure that is signed and verified as authentication credentials.",
"type": "object",
"required": [
"cnonce",
"epk",
"nonce"
],
"properties": {
"cnonce": {
"$ref": "#/definitions/Nonce"
},
"epk": {
"$ref": "#/definitions/RequestVerifier"
},
"nonce": {
"$ref": "#/definitions/Nonce"
}
},
"definitions": {
"Nonce": {
"description": "A unique random string. Authentication credentials have two of these: one generated by the server, and one by the client. This structure is agnostic to the syntax of the string.",
"type": "string"
},
"RequestVerifier": {
"description": "The server half of an ephemeral request-signing key.",
"type": "string"
}
}
}
Loading
Loading