Storage locker and execution boundary

api/src/lib.rs

@@ -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(),
         }
     }
 }

client/src/cli.rs

@@ -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;
             }
         };
@@ -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}"),
         }
     }
 
@@ -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}");
         }
     }
 }
@@ -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}"),
     }
 }
 
@@ -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}");
         }
     }
 
@@ -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}"
+                            )
+                        }
                     }
                 }
             }

client/src/commands.rs

@@ -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};
@@ -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 {
@@ -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}")]

client/src/tunnel.rs

@@ -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}");
                             }
                         });
                     }

common/src/authn.rs

@@ -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;
 
 }
@@ -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,
@@ -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() {

common/src/jobs.rs

@@ -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,
         })
     }
 }
@@ -153,6 +154,7 @@ pub enum LastJob {
     None,
     Some(SignedJob),
     Burned(JobId),
+    Resumed(JobId),
 }
 
 #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
@@ -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)
@@ -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.
@@ -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>;
@@ -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 { .. }
         )
     }
 
@@ -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 {
@@ -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,
         }
     }
 
@@ -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

common/src/keys.rs

@@ -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)
     }

common/tests/output/challenge-response-schema.json

@@ -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"
+    }
+  }
+}