Skip to main content

rust_web_server/fleet/
mod.rs

1//! Fleet config pull (18.23.0, #80): many instances, one source of truth.
2//!
3//! With `[fleet] source` set, rws fetches its config file from an HTTPS
4//! URL, an S3 object or a git repository every `interval_secs`, checks its
5//! Ed25519 signature (`<source>.sig`) against `public_key`, validates it
6//! like `rws --check`, writes it over the local config file and reloads.
7//! A config that fails any step is refused and the running one stays;
8//! `report_url` gets each outcome. `{node}` in the source is the node name
9//! (`[cluster] node`, else the host name), so each instance can have its
10//! own file.
11//!
12//! The fleet config is the declared layer; runtime changes made through
13//! the admin API ([`crate::cluster`]) are the live one.
14//!
15//! `rws fleet keygen` makes a key pair; `rws fleet sign <file> --key <key>`
16//! writes `<file>.sig`.
17
18use std::path::PathBuf;
19use std::sync::{Mutex, OnceLock};
20use std::time::Duration;
21
22use crate::core::base64::Base64;
23use crate::core::json::escape as json_escape;
24
25pub const RWS_CONFIG_FLEET_SOURCE: &str = "RWS_CONFIG_FLEET_SOURCE";
26pub const RWS_CONFIG_FLEET_INTERVAL_SECS: &str = "RWS_CONFIG_FLEET_INTERVAL_SECS";
27pub const RWS_CONFIG_FLEET_PUBLIC_KEY: &str = "RWS_CONFIG_FLEET_PUBLIC_KEY";
28pub const RWS_CONFIG_FLEET_INSECURE: &str = "RWS_CONFIG_FLEET_INSECURE";
29pub const RWS_CONFIG_FLEET_REPORT_URL: &str = "RWS_CONFIG_FLEET_REPORT_URL";
30pub const RWS_CONFIG_FLEET_S3_ENDPOINT: &str = "RWS_CONFIG_FLEET_S3_ENDPOINT";
31
32/// Where the config comes from.
33#[derive(Debug, Clone, PartialEq)]
34pub enum Source {
35    Http(String),
36    S3 { bucket: String, key: String },
37    Git { repo: String, branch: String, path: String },
38}
39
40impl Source {
41    /// `https://…`, `s3://bucket/key`, `git+https://host/repo.git#branch:path`
42    /// (`git+ssh://`, `git+file://` too); `{node}` replaced.
43    pub fn parse(text: &str, node: &str) -> Result<Source, String> {
44        let text = text.trim().replace("{node}", node);
45        if text.starts_with("https://") || text.starts_with("http://") {
46            return Ok(Source::Http(text));
47        }
48        if let Some(rest) = text.strip_prefix("s3://") {
49            let (bucket, key) = rest.split_once('/').ok_or_else(|| format!("[fleet] source {text:?}: expected s3://bucket/key"))?;
50            if bucket.is_empty() || key.is_empty() {
51                return Err(format!("[fleet] source {text:?}: expected s3://bucket/key"));
52            }
53            return Ok(Source::S3 { bucket: bucket.to_string(), key: key.to_string() });
54        }
55        if let Some(rest) = text.strip_prefix("git+") {
56            let (repo, fragment) = rest.split_once('#').ok_or_else(|| format!("[fleet] source {text:?}: expected git+<url>#<branch>:<path>"))?;
57            let (branch, path) = fragment.split_once(':').ok_or_else(|| format!("[fleet] source {text:?}: expected git+<url>#<branch>:<path>"))?;
58            if branch.is_empty() || path.is_empty() || branch.starts_with('-') {
59                return Err(format!("[fleet] source {text:?}: expected git+<url>#<branch>:<path>"));
60            }
61            return Ok(Source::Git { repo: repo.to_string(), branch: branch.to_string(), path: path.trim_start_matches('/').to_string() });
62        }
63        Err(format!("[fleet] source {text:?}: expected https://…, s3://bucket/key or git+<url>#<branch>:<path>"))
64    }
65}
66
67/// `[fleet]` settings.
68#[derive(Debug, Clone)]
69pub struct FleetConfig {
70    pub source: Source,
71    pub interval: Duration,
72    /// Ed25519 public key (32 bytes); `None` only with `insecure = true`.
73    pub public_key: Option<Vec<u8>>,
74    pub report_url: String,
75    pub node: String,
76    /// The local config file written and reloaded.
77    pub config_path: PathBuf,
78    /// Where a git source is cloned.
79    pub state_dir: PathBuf,
80    pub s3_endpoint: String,
81}
82
83/// A public key: base64 of the 32 raw bytes, inline or in a file.
84pub fn read_public_key(value: &str) -> Result<Vec<u8>, String> {
85    let value = value.trim();
86    let text = if value.len() != 44 && std::path::Path::new(value).is_file() {
87        std::fs::read_to_string(value).map_err(|e| format!("[fleet] public_key {value}: {e}"))?
88    } else {
89        value.to_string()
90    };
91    let key = Base64::decode(text.trim().to_string()).map_err(|e| format!("[fleet] public_key: not base64: {e}"))?;
92    if key.len() != 32 {
93        return Err(format!("[fleet] public_key: an Ed25519 public key is 32 bytes, this one is {}", key.len()));
94    }
95    Ok(key)
96}
97
98impl FleetConfig {
99    /// `None` without `[fleet] source`.
100    pub fn from_env() -> Option<Result<FleetConfig, String>> {
101        let env = |name: &str| std::env::var(name).ok().map(|v| v.trim().to_string()).filter(|v| !v.is_empty());
102        let source = env(RWS_CONFIG_FLEET_SOURCE)?;
103        Some((|| {
104            let node = crate::admin::access::node_name();
105            let insecure = env(RWS_CONFIG_FLEET_INSECURE).is_some_and(|v| v.eq_ignore_ascii_case("true"));
106            let public_key = match env(RWS_CONFIG_FLEET_PUBLIC_KEY) {
107                Some(k) => Some(read_public_key(&k)?),
108                None if insecure => None,
109                None => return Err("[fleet] needs public_key (or insecure = true to skip signatures)".to_string()),
110            };
111            Ok(FleetConfig {
112                source: Source::parse(&source, &node)?,
113                interval: Duration::from_secs(env(RWS_CONFIG_FLEET_INTERVAL_SECS).and_then(|v| v.parse().ok()).unwrap_or(30).max(5)),
114                public_key,
115                report_url: env(RWS_CONFIG_FLEET_REPORT_URL).unwrap_or_default(),
116                node,
117                config_path: PathBuf::from(env("RWS_CONFIG_FILE").unwrap_or_else(|| "rws.config.toml".to_string())),
118                state_dir: PathBuf::from(".rws/fleet"),
119                s3_endpoint: env(RWS_CONFIG_FLEET_S3_ENDPOINT).unwrap_or_default(),
120            })
121        })())
122    }
123}
124
125/// What a fetch brought.
126#[derive(Debug, PartialEq)]
127pub enum Fetched {
128    /// The source says nothing changed (HTTP 304).
129    NotModified,
130    Config { body: Vec<u8>, signature: Option<Vec<u8>> },
131}
132
133/// The signature file `rws fleet sign` writes (18.24.0): one line,
134/// `rws-fleet-sig v2 <serial> <base64 signature>`, the signature over
135/// [`signed_message`]. The serial only grows: an instance refuses a config
136/// whose serial isn't above the one it applied, so an older signed file
137/// can't roll it back.
138pub const SIG_PREFIX: &str = "rws-fleet-sig v2 ";
139
140/// What a v2 signature covers: a domain tag, the serial, then the file.
141pub fn signed_message(serial: u64, body: &[u8]) -> Vec<u8> {
142    let mut m = format!("rws-fleet-v2\n{serial}\n").into_bytes();
143    m.extend_from_slice(body);
144    m
145}
146
147/// A parsed signature file: the serial (0 for the 18.23 format, which
148/// signed the body alone) and the signature bytes.
149pub fn parse_signature(raw: &[u8]) -> Result<(u64, Vec<u8>), String> {
150    let text = String::from_utf8_lossy(raw);
151    if let Some(rest) = text.trim().strip_prefix(SIG_PREFIX) {
152        let (serial, sig) = rest.split_once(' ').ok_or("the signature file is `rws-fleet-sig v2 <serial> <signature>`")?;
153        let serial: u64 = serial.parse().map_err(|_| format!("the signature's serial {serial:?} isn't a number"))?;
154        return Ok((serial, signature_bytes(sig.trim().as_bytes())?));
155    }
156    Ok((0, signature_bytes(raw)?))
157}
158
159/// A signature: base64, or the 64 raw bytes.
160fn signature_bytes(raw: &[u8]) -> Result<Vec<u8>, String> {
161    if raw.len() == 64 {
162        return Ok(raw.to_vec());
163    }
164    // 64 bytes are 88 base64 characters: anything much longer isn't one
165    // (and isn't worth decoding).
166    if raw.len() > 256 {
167        return Err(format!("an Ed25519 signature is 64 bytes; this is {} characters", raw.len()));
168    }
169    let sig = Base64::decode(String::from_utf8_lossy(raw).trim().to_string()).map_err(|e| format!("the signature isn't base64: {e}"))?;
170    if sig.len() != 64 {
171        return Err(format!("an Ed25519 signature is 64 bytes, this one is {}", sig.len()));
172    }
173    Ok(sig)
174}
175
176/// Check `body` against a signature file with `public_key`; the serial it
177/// carries (0 for an 18.23-format file).
178pub fn verify(body: &[u8], signature: &[u8], public_key: &[u8]) -> Result<u64, String> {
179    let (serial, sig) = parse_signature(signature)?;
180    let message = if serial == 0 && !String::from_utf8_lossy(signature).trim_start().starts_with(SIG_PREFIX) { body.to_vec() } else { signed_message(serial, body) };
181    aws_lc_rs::signature::UnparsedPublicKey::new(&aws_lc_rs::signature::ED25519, public_key)
182        .verify(&message, &sig)
183        .map_err(|_| "the signature doesn't match (wrong key, or the file was changed after signing)".to_string())?;
184    Ok(serial)
185}
186
187/// The `[fleet]` settings a config file sets (`RWS_CONFIG_FLEET_*` → value).
188fn fleet_settings(text: &str) -> Vec<(String, String)> {
189    let mut v: Vec<(String, String)> = crate::entry_point::config_file::collect_settings(std::io::Cursor::new(text.as_bytes()), String::new())
190        .map(|c| c.settings)
191        .unwrap_or_default()
192        .into_iter()
193        .filter(|(k, _)| k.starts_with("RWS_CONFIG_FLEET_"))
194        .map(|(k, v)| (k, v.trim().to_string()))
195        .collect();
196    v.sort();
197    v
198}
199
200fn sha256_hex(bytes: &[u8]) -> String {
201    crate::core::sha256::sha256(bytes).iter().map(|b| format!("{b:02x}")).collect()
202}
203
204/// The puller: fetch, verify, check, write, reload, report.
205pub struct Fleet {
206    config: FleetConfig,
207    state: Mutex<FleetState>,
208}
209
210#[derive(Debug, Default, Clone)]
211struct FleetState {
212    etag: Option<String>,
213    applied: Option<String>,
214    /// The serial of the applied config (persisted in the state directory).
215    serial: u64,
216    last_check: u64,
217    last_error: Option<String>,
218}
219
220/// One round's outcome.
221#[derive(Debug, PartialEq)]
222pub enum Outcome {
223    Unchanged,
224    Applied(String),
225    Refused(String),
226}
227
228fn unix_now() -> u64 {
229    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
230}
231
232impl Fleet {
233    pub fn new(config: FleetConfig) -> Fleet {
234        let applied = std::fs::read(&config.config_path).ok().map(|b| sha256_hex(&b));
235        let serial = std::fs::read_to_string(config.state_dir.join("serial")).ok().and_then(|s| s.trim().parse().ok()).unwrap_or(0);
236        Fleet { config, state: Mutex::new(FleetState { applied, serial, ..FleetState::default() }) }
237    }
238
239    fn fetch(&self) -> Result<Fetched, String> {
240        let want_sig = self.config.public_key.is_some();
241        match &self.config.source {
242            Source::Http(url) => {
243                let etag = self.state.lock().unwrap_or_else(|e| e.into_inner()).etag.clone();
244                let client = crate::http_client::Client::new().timeout_ms(15_000);
245                let mut request = client.get(url);
246                if let Some(tag) = &etag {
247                    request = request.header("If-None-Match", tag);
248                }
249                let response = request.send().map_err(|e| format!("fetching {url}: {e}"))?;
250                if response.status() == 304 {
251                    return Ok(Fetched::NotModified);
252                }
253                if !response.is_success() {
254                    return Err(format!("fetching {url}: HTTP {}", response.status()));
255                }
256                let new_etag = response.header("ETag").map(str::to_string);
257                let body = response.bytes().to_vec();
258                let signature = if want_sig {
259                    let sig = client.get(&format!("{url}.sig")).send().map_err(|e| format!("fetching {url}.sig: {e}"))?;
260                    if !sig.is_success() {
261                        return Err(format!("fetching {url}.sig: HTTP {}", sig.status()));
262                    }
263                    Some(sig.bytes().to_vec())
264                } else {
265                    None
266                };
267                self.state.lock().unwrap_or_else(|e| e.into_inner()).etag = new_etag;
268                Ok(Fetched::Config { body, signature })
269            }
270            #[cfg(feature = "storage-s3")]
271            Source::S3 { bucket, key } => {
272                use crate::storage::Storage;
273                let region = std::env::var("AWS_REGION").or_else(|_| std::env::var("AWS_DEFAULT_REGION")).unwrap_or_else(|_| "us-east-1".to_string());
274                let endpoint = if self.config.s3_endpoint.is_empty() { format!("https://s3.{region}.amazonaws.com") } else { self.config.s3_endpoint.clone() };
275                let s3 = crate::storage::S3Storage::new(crate::storage::S3Config {
276                    bucket: bucket.clone(),
277                    region,
278                    access_key: std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_default(),
279                    secret_key: std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_default(),
280                    endpoint,
281                });
282                let body = s3.get(key).map_err(|e| format!("s3://{bucket}/{key}: {e}"))?;
283                let signature = if want_sig { Some(s3.get(&format!("{key}.sig")).map_err(|e| format!("s3://{bucket}/{key}.sig: {e}"))?) } else { None };
284                Ok(Fetched::Config { body, signature })
285            }
286            #[cfg(not(feature = "storage-s3"))]
287            Source::S3 { .. } => Err("s3:// sources need the storage-s3 feature, which this build doesn't include".to_string()),
288            Source::Git { repo, branch, path } => {
289                let dir = self.config.state_dir.join("git");
290                let git = |args: &[&str]| -> Result<(), String> {
291                    let out = std::process::Command::new("git").args(args).output().map_err(|e| format!("running git: {e} (is git installed?)"))?;
292                    if out.status.success() {
293                        Ok(())
294                    } else {
295                        Err(format!("git {}: {}", args.first().unwrap_or(&""), String::from_utf8_lossy(&out.stderr).trim()))
296                    }
297                };
298                let d = dir.to_string_lossy().to_string();
299                if !dir.join(".git").is_dir() {
300                    std::fs::create_dir_all(&self.config.state_dir).map_err(|e| e.to_string())?;
301                    git(&["clone", "--quiet", "--depth", "1", "--branch", branch, "--", repo, &d])?;
302                } else {
303                    git(&["-C", &d, "fetch", "--quiet", "--depth", "1", "origin", branch])?;
304                    git(&["-C", &d, "reset", "--quiet", "--hard", "FETCH_HEAD"])?;
305                }
306                let file = dir.join(path);
307                if !file.canonicalize().is_ok_and(|f| f.starts_with(dir.canonicalize().unwrap_or_default())) {
308                    return Err(format!("{path} isn't in the repository"));
309                }
310                let body = std::fs::read(&file).map_err(|e| format!("{path}: {e}"))?;
311                let signature = if want_sig { Some(std::fs::read(dir.join(format!("{path}.sig"))).map_err(|e| format!("{path}.sig: {e}"))?) } else { None };
312                Ok(Fetched::Config { body, signature })
313            }
314        }
315    }
316
317    /// One round.
318    pub fn tick(&self) -> Outcome {
319        let outcome = match self.fetch() {
320            Ok(Fetched::NotModified) => Outcome::Unchanged,
321            Ok(Fetched::Config { body, signature }) => self.apply(body, signature),
322            Err(e) => Outcome::Refused(e),
323        };
324        {
325            let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
326            state.last_check = unix_now();
327            match &outcome {
328                Outcome::Refused(e) => {
329                    if state.last_error.as_deref() != Some(e) {
330                        eprintln!("[fleet] {e}; the running config stays");
331                    }
332                    state.last_error = Some(e.clone());
333                }
334                Outcome::Applied(hash) => {
335                    println!("[fleet] applied config {}", &hash[..12]);
336                    state.last_error = None;
337                }
338                Outcome::Unchanged => state.last_error = None,
339            }
340        }
341        if !matches!(outcome, Outcome::Unchanged) {
342            self.report(&outcome);
343        }
344        outcome
345    }
346
347    fn apply(&self, body: Vec<u8>, signature: Option<Vec<u8>>) -> Outcome {
348        let hash = sha256_hex(&body);
349        let (applied, applied_serial) = {
350            let state = self.state.lock().unwrap_or_else(|e| e.into_inner());
351            (state.applied.clone(), state.serial)
352        };
353        let mut serial = 0;
354        if let Some(key) = &self.config.public_key {
355            let Some(sig) = signature else { return Outcome::Refused("no signature".into()) };
356            match verify(&body, &sig, key) {
357                Ok(s) => serial = s,
358                Err(e) => return Outcome::Refused(e),
359            }
360            if applied.as_deref() == Some(hash.as_str()) {
361                return Outcome::Unchanged;
362            }
363            // Rollback: a validly signed but older file.
364            if applied_serial > 0 && serial <= applied_serial {
365                return Outcome::Refused(format!("the config's serial {serial} isn't above the applied {applied_serial} (an older signed file?)"));
366            }
367        } else if applied.as_deref() == Some(hash.as_str()) {
368            return Outcome::Unchanged;
369        }
370        let Ok(text) = String::from_utf8(body.clone()) else { return Outcome::Refused("the config isn't UTF-8".into()) };
371        let path = &self.config.config_path;
372        let previous = std::fs::read(path).ok();
373        // The fleet settings themselves (source, key, insecure) can't be
374        // changed through the fleet: that would let one signed file turn off
375        // signatures or move the source at the next restart.
376        let current = previous.as_deref().map(|p| fleet_settings(&String::from_utf8_lossy(p))).unwrap_or_default();
377        if fleet_settings(&text) != current {
378            return Outcome::Refused("the fetched config changes [fleet] settings; change them on the instance (or its environment), not through the fleet".to_string());
379        }
380        let report = crate::config_check::check_str(&text);
381        if !report.is_ok() {
382            return Outcome::Refused(format!("the fetched config doesn't pass rws --check: {}", report.errors.join("; ")));
383        }
384        let write = |bytes: &[u8]| {
385            let tmp = path.with_extension("fleet.tmp");
386            std::fs::write(&tmp, bytes).and_then(|_| std::fs::rename(&tmp, path))
387        };
388        if let Err(e) = write(&body) {
389            return Outcome::Refused(format!("writing {}: {e}", path.display()));
390        }
391        if let Err(e) = self.reload() {
392            // Put the previous file back, so a restart doesn't load the
393            // config the running server refused.
394            if let Some(prev) = &previous {
395                if let Err(w) = write(prev) {
396                    return Outcome::Refused(format!("{e}; restoring {} failed: {w}", path.display()));
397                }
398                let _ = self.reload();
399            }
400            return Outcome::Refused(e);
401        }
402        let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
403        state.applied = Some(hash.clone());
404        if serial > 0 {
405            state.serial = serial;
406            let _ = std::fs::create_dir_all(&self.config.state_dir).and_then(|_| std::fs::write(self.config.state_dir.join("serial"), serial.to_string()));
407        }
408        Outcome::Applied(hash)
409    }
410
411    /// Reload as SIGHUP does and wait for it; `Err` with the errors it
412    /// reported. A no-op before the server is up.
413    fn reload(&self) -> Result<(), String> {
414        if !crate::metrics::SERVER_READY.load(std::sync::atomic::Ordering::SeqCst) {
415            return Ok(());
416        }
417        let (before, _) = crate::config_reload::last_reload();
418        crate::config_reload::request_reload();
419        let deadline = std::time::Instant::now() + Duration::from_secs(10);
420        while std::time::Instant::now() < deadline {
421            let (count, errors) = crate::config_reload::last_reload();
422            if count > before {
423                return if errors.is_empty() { Ok(()) } else { Err(format!("the reload reported: {}", errors.join("; "))) };
424            }
425            std::thread::sleep(Duration::from_millis(50));
426        }
427        Ok(())
428    }
429
430    fn report(&self, outcome: &Outcome) {
431        if self.config.report_url.is_empty() {
432            return;
433        }
434        let (applied, detail) = match outcome {
435            Outcome::Applied(hash) => (true, format!(r#""config":"{hash}""#)),
436            Outcome::Refused(e) => (false, format!(r#""error":"{}""#, json_escape(e))),
437            Outcome::Unchanged => return,
438        };
439        let body = format!(r#"{{"node":"{}","rws":"{}","applied":{applied},{detail},"at":{}}}"#, json_escape(&self.config.node), env!("CARGO_PKG_VERSION"), unix_now());
440        if let Err(e) = crate::http_client::Client::new().timeout_ms(10_000).post(&self.config.report_url).body_json(&body).send() {
441            eprintln!("[fleet] report to {}: {e}", self.config.report_url);
442        }
443    }
444
445    /// `{"source":…,"node":…,"config":…,"last_check":…,"error":…}`.
446    pub fn status_json(&self) -> String {
447        let state = self.state.lock().unwrap_or_else(|e| e.into_inner()).clone();
448        let source = match &self.config.source {
449            Source::Http(u) => u.clone(),
450            Source::S3 { bucket, key } => format!("s3://{bucket}/{key}"),
451            Source::Git { repo, branch, path } => format!("git+{repo}#{branch}:{path}"),
452        };
453        let opt = |v: &Option<String>| v.as_ref().map_or("null".to_string(), |s| format!("\"{}\"", json_escape(s)));
454        format!(
455            r#"{{"source":"{}","node":"{}","config":{},"last_check":{},"error":{}}}"#,
456            json_escape(&source),
457            json_escape(&self.config.node),
458            opt(&state.applied),
459            state.last_check,
460            opt(&state.last_error)
461        )
462    }
463}
464
465fn global() -> &'static OnceLock<Fleet> {
466    static FLEET: OnceLock<Fleet> = OnceLock::new();
467    &FLEET
468}
469
470/// The fleet status for the admin API, or `None` without `[fleet]`.
471pub fn status_json() -> Option<String> {
472    global().get().map(Fleet::status_json)
473}
474
475/// The binary's entry point: with `[fleet] source`, pull on a thread.
476pub fn start_from_env() {
477    let config = match FleetConfig::from_env() {
478        None => return,
479        Some(Ok(c)) => c,
480        Some(Err(e)) => return eprintln!("[fleet] not started: {e}"),
481    };
482    let interval = config.interval;
483    if global().set(Fleet::new(config)).is_err() {
484        return;
485    }
486    println!("[fleet] pulling the config every {}s", interval.as_secs());
487    let _ = std::thread::Builder::new().name("rws-fleet".into()).spawn(move || loop {
488        if let Some(fleet) = global().get() {
489            fleet.tick();
490        }
491        std::thread::sleep(interval);
492    });
493}
494
495// ── rws fleet keygen / sign ──────────────────────────────────────────────────
496
497/// `rws fleet keygen [<prefix>]` / `rws fleet sign <file> --key <file>`:
498/// `Some(exit code)` when `args` (a full argv) asks for one.
499pub fn command(args: &[String]) -> Option<i32> {
500    if args.get(1).map(String::as_str) != Some("fleet") {
501        return None;
502    }
503    let rest: Vec<&str> = args[2..].iter().map(String::as_str).collect();
504    let usage = "Usage: rws fleet keygen [<prefix>]   (writes <prefix>.key and <prefix>.pub; default fleet)\n       rws fleet sign <file> --key <prefix>.key [--serial N]   (writes <file>.sig; serial default: now)";
505    let result = match rest.as_slice() {
506        ["keygen"] => keygen("fleet"),
507        ["keygen", prefix] => keygen(prefix),
508        ["sign", file, "--key", key] | ["sign", "--key", key, file] => sign(file, key, None),
509        ["sign", file, "--key", key, "--serial", serial] => match serial.parse() {
510            Ok(n) if n > 0 => sign(file, key, Some(n)),
511            _ => Err(format!("--serial {serial:?}: expected a number above 0")),
512        },
513        _ => {
514            eprintln!("{usage}");
515            return Some(2);
516        }
517    };
518    Some(match result {
519        Ok(message) => {
520            println!("{message}");
521            0
522        }
523        Err(e) => {
524            eprintln!("rws fleet: {e}");
525            1
526        }
527    })
528}
529
530fn keygen(prefix: &str) -> Result<String, String> {
531    use aws_lc_rs::signature::KeyPair;
532    let rng = aws_lc_rs::rand::SystemRandom::new();
533    let pkcs8 = aws_lc_rs::signature::Ed25519KeyPair::generate_pkcs8(&rng).map_err(|_| "key generation failed".to_string())?;
534    let pair = aws_lc_rs::signature::Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).map_err(|_| "key generation failed".to_string())?;
535    let private = Base64::encode(pkcs8.as_ref())?;
536    let public = Base64::encode(pair.public_key().as_ref())?;
537    let key_path = format!("{prefix}.key");
538    let mut options = std::fs::OpenOptions::new();
539    options.write(true).create_new(true);
540    #[cfg(unix)]
541    std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
542    let mut file = options.open(&key_path).map_err(|e| format!("{key_path}: {e}"))?;
543    std::io::Write::write_all(&mut file, format!("{private}\n").as_bytes()).map_err(|e| e.to_string())?;
544    std::fs::write(format!("{prefix}.pub"), format!("{public}\n")).map_err(|e| format!("{prefix}.pub: {e}"))?;
545    Ok(format!("wrote {key_path} (keep it secret) and {prefix}.pub\npublic_key = \"{public}\""))
546}
547
548/// Sign `file` with serial `serial`, else the current Unix time (and above
549/// the serial of an existing `<file>.sig`).
550fn sign(file: &str, key: &str, serial: Option<u64>) -> Result<String, String> {
551    let body = std::fs::read(file).map_err(|e| format!("{file}: {e}"))?;
552    let pkcs8 = Base64::decode(std::fs::read_to_string(key).map_err(|e| format!("{key}: {e}"))?.trim().to_string()).map_err(|e| format!("{key}: not a key from rws fleet keygen: {e}"))?;
553    let pair = aws_lc_rs::signature::Ed25519KeyPair::from_pkcs8(&pkcs8).map_err(|_| format!("{key}: not a key from rws fleet keygen"))?;
554    let report = crate::config_check::check_str_offline(&String::from_utf8_lossy(&body));
555    if !report.is_ok() {
556        return Err(format!("{file} doesn't pass rws --check: {}", report.errors.join("; ")));
557    }
558    // Above the file's previous signature too, so two signings within a
559    // second still grow.
560    let previous = std::fs::read(format!("{file}.sig")).ok().and_then(|raw| parse_signature(&raw).ok()).map_or(0, |(n, _)| n);
561    let serial = serial.unwrap_or_else(|| unix_now().max(previous + 1));
562    let signature = Base64::encode(pair.sign(&signed_message(serial, &body)).as_ref())?;
563    std::fs::write(format!("{file}.sig"), format!("{SIG_PREFIX}{serial} {signature}\n")).map_err(|e| format!("{file}.sig: {e}"))?;
564    Ok(format!("wrote {file}.sig (serial {serial})"))
565}
566
567#[cfg(test)]
568mod tests;