1use 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#[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 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#[derive(Debug, Clone)]
69pub struct FleetConfig {
70 pub source: Source,
71 pub interval: Duration,
72 pub public_key: Option<Vec<u8>>,
74 pub report_url: String,
75 pub node: String,
76 pub config_path: PathBuf,
78 pub state_dir: PathBuf,
80 pub s3_endpoint: String,
81}
82
83pub 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 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#[derive(Debug, PartialEq)]
127pub enum Fetched {
128 NotModified,
130 Config { body: Vec<u8>, signature: Option<Vec<u8>> },
131}
132
133pub const SIG_PREFIX: &str = "rws-fleet-sig v2 ";
139
140pub 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
147pub 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
159fn signature_bytes(raw: &[u8]) -> Result<Vec<u8>, String> {
161 if raw.len() == 64 {
162 return Ok(raw.to_vec());
163 }
164 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
176pub 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
187fn 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
204pub 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 serial: u64,
216 last_check: u64,
217 last_error: Option<String>,
218}
219
220#[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 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 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 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 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 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 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
470pub fn status_json() -> Option<String> {
472 global().get().map(Fleet::status_json)
473}
474
475pub 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
495pub 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
548fn 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 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;