1use std::fs;
23use std::path::{Path, PathBuf};
24use std::process::Command;
25
26use crate::builtins::JoinConfigArgs;
27use crate::context::home_dir;
28use crate::style;
29use crate::util::platform;
30use crate::util::prompt;
31use crate::util::system;
32
33const KEY_NAME: &str = "flodl-join";
35const WORKER_KEY_PATH: &str = "~/.ssh/flodl-join";
39const SERVED_SUBDIR: &str = ".flodl/run";
42
43pub fn run(cli: &JoinConfigArgs) -> i32 {
44 match wizard(cli) {
45 Ok(report) => {
46 if cli.json {
47 println!(
48 "{}",
49 serde_json::to_string_pretty(&report.to_json())
50 .expect("a report value serializes"),
51 );
52 } else {
53 print!("{}", report.render_human());
54 }
55 0
56 }
57 Err(e) => {
58 crate::cli_error!("{e}");
59 1
60 }
61 }
62}
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
68enum Door {
69 B,
72 A,
75 Nologin,
78}
79
80impl Door {
81 fn parse(s: Option<&str>) -> Result<Self, String> {
82 match s.unwrap_or("b") {
83 "b" | "B" => Ok(Door::B),
84 "a" | "A" => Ok(Door::A),
85 "nologin" => Ok(Door::Nologin),
86 "c" | "C" => Err("door `c` (a second, source-only key) serves cross-host \
87 routing that fdl join's single identity cannot express — \
88 compose it manually from the guide's recipe \
89 (docs/ddp/02-cluster-guide.md), or use door `b`"
90 .to_string()),
91 other => Err(format!(
92 "unknown door `{other}` — one of `b` (rrsync source pull, \
93 the publish-then-join default), `a` (sftp data mount), \
94 `nologin` (tunnel only)"
95 )),
96 }
97 }
98}
99
100struct Report {
103 label: String,
104 farm_dir: PathBuf,
105 overlay_path: PathBuf,
106 overlay_action: OverlayAction,
107 key_path: PathBuf,
108 pub_line: String,
109 key_action: KeyAction,
110 reuse_warning: Option<String>,
111 authorized_line: String,
112 match_block: String,
113 door: Door,
114 worker_yml_path: PathBuf,
115 worker_yml: String,
116 publish_block: Option<String>,
117 bin_caveat: Option<String>,
118 freshness: Option<String>,
119 notes_path: PathBuf,
120 controller: Endpoint,
121 install: InstallAction,
122 cloud_init_path: Option<PathBuf>,
123 checks: Vec<Check>,
125 sshd_conf_path: PathBuf,
127 plat: platform::Platform,
128 in_container: bool,
132 docker_services: Vec<String>,
135}
136
137#[derive(Debug, Clone, Copy, PartialEq, Eq)]
138enum OverlayAction {
139 Scaffolded,
140 TokenReplaced,
141 TokenReused,
142 SnippetPrinted,
146}
147
148#[derive(Debug, Clone, Copy, PartialEq, Eq)]
149enum KeyAction {
150 Generated,
151 Regenerated,
152 Reused,
153}
154
155#[derive(Debug, Clone)]
157struct Endpoint {
158 user: String,
159 host: String,
160 port: u16,
161}
162
163impl Endpoint {
164 fn parse(spec: Option<&str>) -> Result<Self, String> {
165 let default_user = std::env::var("USER")
166 .or_else(|_| std::env::var("USERNAME"))
167 .unwrap_or_else(|_| "op".to_string());
168 let Some(spec) = spec else {
169 return Ok(Endpoint {
170 user: default_user,
171 host: crate::cluster::resolve_local_hostname(),
172 port: 22,
173 });
174 };
175 let (user, rest) = match spec.split_once('@') {
176 Some((u, r)) if !u.is_empty() => (u.to_string(), r),
177 Some(_) => return Err(format!("--controller `{spec}`: empty user before `@`")),
178 None => (default_user, spec),
179 };
180 let (host, port) = match rest.rsplit_once(':') {
181 Some((h, p)) => (
182 h.to_string(),
183 p.parse::<u16>()
184 .map_err(|_| format!("--controller `{spec}`: bad port `{p}`"))?,
185 ),
186 None => (rest.to_string(), 22),
187 };
188 if host.is_empty() {
189 return Err(format!("--controller `{spec}`: empty host"));
190 }
191 Ok(Endpoint { user, host, port })
192 }
193}
194
195fn wizard(cli: &JoinConfigArgs) -> Result<Report, String> {
196 let label = resolve_label(cli)?;
197 validate_label(&label)?;
198 let cwd = std::env::current_dir().map_err(|e| format!("cannot read cwd: {e}"))?;
206 let root = match crate::config::find_project_config(&cwd) {
207 Some(p) => p
208 .parent()
209 .map(Path::to_path_buf)
210 .unwrap_or_else(|| cwd.clone()),
211 None => {
212 let target = cwd.join("fdl.yml");
213 if !confirm(
214 cli,
215 &format!(
216 "no fdl.yml here — create a minimal one at {} so the farm \
217 overlay has a base?",
218 target.display()
219 ),
220 )? {
221 return Err("a farm overlay needs a base fdl.yml to merge onto".into());
222 }
223 fs::write(
224 &target,
225 "# fdl.yml — created by `fdl join-config` so farm overlays\n\
226 # (fdl.<label>.yml) have a base to merge onto. Project\n\
227 # config goes here as it grows.\n",
228 )
229 .map_err(|e| format!("cannot write {}: {e}", target.display()))?;
230 cwd.clone()
231 }
232 };
233
234 let farm_dir = root.join(".fdl").join(&label);
235 let prior = recover_shape(&farm_dir);
239 let door = match cli.door.as_deref() {
240 Some(d) => Door::parse(Some(d))?,
241 None => prior.as_ref().map(|p| p.0).unwrap_or(Door::B),
242 };
243 let controller = match cli.controller.as_deref() {
244 Some(c) => Endpoint::parse(Some(c))?,
245 None => match prior.as_ref().map(|p| p.1.clone()) {
246 Some(spec) => Endpoint::parse(Some(&spec))?,
247 None => Endpoint::parse(None)?,
248 },
249 };
250 let keys_dir = farm_dir.join("keys");
251 fs::create_dir_all(&keys_dir)
252 .map_err(|e| format!("cannot create {}: {e}", keys_dir.display()))?;
253 let self_ignore = root.join(".fdl").join(".gitignore");
256 if !self_ignore.is_file() {
257 fs::write(&self_ignore, "*\n")
258 .map_err(|e| format!("cannot write {}: {e}", self_ignore.display()))?;
259 }
260
261 let key_path = keys_dir.join(KEY_NAME);
263 let key_action = ensure_key(cli, &key_path, &label)?;
264 let pub_line = fs::read_to_string(key_path.with_extension("pub"))
265 .map_err(|e| format!("cannot read the generated public key: {e}"))?
266 .trim()
267 .to_string();
268
269 let overlay_path = root.join(format!("fdl.{label}.yml"));
271 let cmd_hint = command_hint(match &cli.crate_dir {
276 Some(d) => {
277 let p = PathBuf::from(d);
278 if p.is_absolute() { p } else { cwd.join(p) }
279 }
280 None => cwd.clone(),
281 });
282 let (token, overlay_action) =
283 ensure_overlay(cli, &overlay_path, &label, &root, cmd_hint.as_deref())?;
284
285 let reuse_warning = foreign_identity_warning(&root, &label, &farm_dir);
289
290 let served_abs = home_dir().join(SERVED_SUBDIR);
292 let authorized_line = authorized_keys_line(door, &served_abs, cli, &pub_line);
293 let match_block = sshd_match_block(&controller.user);
294
295 let crate_dir = match &cli.crate_dir {
297 Some(d) => {
298 let p = PathBuf::from(d);
299 if p.is_absolute() { p } else { cwd.join(p) }
300 }
301 None => cwd.clone(),
302 };
303 let (publish_block, bin_caveat, freshness) = match derive_publish(&crate_dir) {
304 Ok(Some(d)) => (
305 Some(render_publish_block(&d)),
306 d.bin_caveat.clone(),
307 Some(freshness_report(&d.from_root)),
308 ),
309 Ok(None) => (None, None, None),
310 Err(e) => (None, Some(e), None),
311 };
312
313 let worker_yml = render_worker_yml(&label, &controller, &token, door, cli);
315 let worker_yml_path = farm_dir.join("worker.yml");
316 fs::write(&worker_yml_path, &worker_yml)
317 .map_err(|e| format!("cannot write {}: {e}", worker_yml_path.display()))?;
318
319 let notes_path = farm_dir.join("install-notes.md");
321 let notes = render_notes(&label, &authorized_line, &match_block, &controller, door);
322 fs::write(¬es_path, notes)
323 .map_err(|e| format!("cannot write {}: {e}", notes_path.display()))?;
324
325 let plat = platform::Platform::detect();
329 let sshd_conf_path = farm_dir.join(format!("sshd-{label}.conf"));
330 fs::write(
331 &sshd_conf_path,
332 render_sshd_conf(&label, door, controller.port, plat),
333 )
334 .map_err(|e| format!("cannot write {}: {e}", sshd_conf_path.display()))?;
335
336 let checks = preflight(door, controller.port, &served_abs, plat);
338 let services = docker_services(&root, &label);
339
340 let install = install_authorized_line(cli, &authorized_line, controller.port)?;
342
343 let cloud_init_path = if cli.cloud_init {
345 let user = cli.cloud_init_user.as_deref().unwrap_or("ubuntu");
346 let private_key = fs::read_to_string(&key_path)
347 .map_err(|e| format!("cannot read the private key for cloud-init: {e}"))?;
348 let content = render_cloud_init(&label, user, door, &worker_yml, &private_key);
349 let path = farm_dir.join("cloud-init.yml");
350 fs::write(&path, content).map_err(|e| format!("cannot write {}: {e}", path.display()))?;
351 set_mode(&path, 0o600)?;
352 Some(path)
353 } else {
354 None
355 };
356
357 Ok(Report {
358 label,
359 farm_dir,
360 overlay_path,
361 overlay_action,
362 key_path,
363 pub_line,
364 key_action,
365 reuse_warning,
366 authorized_line,
367 match_block,
368 door,
369 worker_yml_path,
370 worker_yml,
371 publish_block,
372 bin_caveat,
373 freshness,
374 notes_path,
375 controller,
376 install,
377 cloud_init_path,
378 checks,
379 sshd_conf_path,
380 plat,
381 in_container: platform::Platform::in_container(),
382 docker_services: services,
383 })
384}
385
386fn home_of(user: &str) -> String {
391 if user == "root" {
392 "/root".to_string()
393 } else {
394 format!("/home/{user}")
395 }
396}
397
398fn render_cloud_init(
420 label: &str,
421 user: &str,
422 door: Door,
423 worker_yml: &str,
424 private_key: &str,
425) -> String {
426 let indent = |s: &str| -> String {
427 s.lines()
428 .map(|l| {
429 if l.is_empty() {
430 String::new()
431 } else {
432 format!(" {l}")
433 }
434 })
435 .collect::<Vec<_>>()
436 .join("\n")
437 };
438 let home = home_of(user);
439
440 let mut packages: Vec<&str> = vec!["curl"];
446 match door {
447 Door::B => packages.extend(["build-essential", "pkg-config", "unzip", "rsync", "git"]),
448 Door::A => packages.push("sshfs"),
449 Door::Nologin => {}
450 }
451 let packages = packages
452 .iter()
453 .map(|p| format!("\x20 - {p}\n"))
454 .collect::<String>();
455
456 let (rust_step, rust_path) = match door {
462 Door::B => (
463 format!(
464 "\x20 - [ sh, -c, \"command -v cargo >/dev/null || \
465 su -l {user} -c 'curl -fsSL https://sh.rustup.rs | \
466 sh -s -- -y --profile minimal --no-modify-path'\" ]\n"
467 ),
468 format!("{home}/.cargo/bin:"),
469 ),
470 _ => (String::new(), String::new()),
471 };
472
473 format!(
474 "#cloud-config\n\
475 # Farm `{label}` worker user-data — generated by `fdl join-config`.\n\
476 # SECRET ARTIFACT: carries the join key and the admission token.\n\
477 # On a provider that bills powered-off instances (DigitalOcean and\n\
478 # the AMD Developer Cloud on top of it), the unit's poweroff stops\n\
479 # the work but NOT the meter: destroy the instance to stop billing.\n\
480 packages:\n{packages}\
481 write_files:\n\
482 \x20 - path: {home}/.ssh/flodl-join\n\
483 \x20 owner: {user}:{user}\n\
484 \x20 permissions: \"0600\"\n\
485 \x20 defer: true\n\
486 \x20 content: |\n{key}\n\
487 \x20 - path: {home}/training/fdl.yml\n\
488 \x20 owner: {user}:{user}\n\
489 \x20 permissions: \"0644\"\n\
490 \x20 defer: true\n\
491 \x20 content: |\n{yml}\n\
492 \x20 - path: /etc/systemd/system/flodl-join.service\n\
493 \x20 permissions: \"0644\"\n\
494 \x20 content: |\n\
495 \x20 [Unit]\n\
496 \x20 Description=flodl walk-in worker (farm {label})\n\
497 \x20 After=network-online.target\n\
498 \x20 Wants=network-online.target\n\
499 \x20 FailureAction=poweroff\n\
500 \n\
501 \x20 [Service]\n\
502 \x20 Type=simple\n\
503 \x20 User={user}\n\
504 \x20 WorkingDirectory={home}/training\n\
505 \x20 Environment=PATH={rust_path}/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n\
506 \x20 ExecStart=/usr/bin/env fdl join\n\
507 \x20 Restart=always\n\
508 \x20 RestartSec=5\n\
509 \x20 RestartPreventExitStatus=2\n\
510 \n\
511 \x20 [Install]\n\
512 \x20 WantedBy=multi-user.target\n\
513 runcmd:\n\
514 \x20 - [ sh, -c, \"command -v fdl >/dev/null || \
515 (curl -fsSL https://flodl.dev/fdl -o /usr/local/bin/fdl && \
516 chmod 0755 /usr/local/bin/fdl)\" ]\n\
517 {rust_step}\
518 \x20 - systemctl daemon-reload\n\
519 \x20 - systemctl enable --now flodl-join.service\n",
520 label = label,
521 user = user,
522 home = home,
523 packages = packages,
524 rust_step = rust_step,
525 rust_path = rust_path,
526 key = indent(private_key),
527 yml = indent(worker_yml),
528 )
529}
530
531fn docker_services(root: &Path, label: &str) -> Vec<String> {
541 let Some(base) = crate::config::find_project_config(root) else {
542 return Vec::new();
543 };
544 let Ok(project) = crate::config::load_project_with_env(&base, Some(label)) else {
545 return Vec::new();
546 };
547 let mut seen: Vec<String> = Vec::new();
548 for spec in project.commands.values() {
549 if let Some(svc) = &spec.docker
550 && !seen.contains(svc)
551 {
552 seen.push(svc.clone());
553 }
554 }
555 seen
556}
557
558#[derive(Debug, Clone)]
563struct Check {
564 what: String,
565 ok: bool,
566 fix: Option<String>,
569}
570
571fn preflight(door: Door, port: u16, served: &Path, plat: platform::Platform) -> Vec<Check> {
580 let mut checks = Vec::new();
581
582 let have_sshd = system::has_command("sshd")
584 || Path::new("/usr/sbin/sshd").exists()
585 || Path::new("/usr/libexec/sshd-keygen-wrapper").exists();
586 checks.push(Check {
587 what: "an ssh daemon on this box".into(),
588 ok: have_sshd,
589 fix: (!have_sshd)
590 .then(|| plat.sshd_package().and_then(|pkg| plat.install(&[pkg])))
591 .flatten(),
592 });
593
594 let listening = sshd_listening(port);
602 let socket_owns = plat == platform::Platform::Debian && port != 22 && socket_activated();
603 checks.push(Check {
604 what: if socket_owns {
605 format!(
606 "something listening on port {port} — ssh.socket owns the \
607 listener, so sshd_config's `Port` is ignored until it is \
608 handed back"
609 )
610 } else {
611 format!("something listening on port {port}")
612 },
613 ok: listening && !socket_owns,
614 fix: (!listening || socket_owns)
615 .then(|| plat.enable_sshd().join(" && "))
616 .filter(|s| !s.is_empty()),
617 });
618
619 if let Some(fix) = plat.allow_ssh_port(port) {
622 checks.push(Check {
623 what: format!("SELinux permits sshd on port {port}"),
624 ok: false,
625 fix: Some(fix),
626 });
627 }
628
629 match door {
631 Door::B => {
632 let have = system::has_command("rrsync");
633 checks.push(Check {
634 what: "`rrsync` (door b runs it as the forced command)".into(),
635 ok: have,
636 fix: (!have).then(|| plat.rrsync_fix()).flatten(),
640 });
641 let served_ok = served.is_dir();
642 checks.push(Check {
643 what: format!("the served directory {} exists", served.display()),
644 ok: served_ok,
645 fix: (!served_ok).then(|| {
646 "fdl publish <source> --bin <artifact> # creates and fills it".to_string()
647 }),
648 });
649 }
650 Door::A => {
651 let have = [
652 "/usr/lib/openssh/sftp-server",
653 "/usr/libexec/openssh/sftp-server",
654 "/usr/libexec/sftp-server",
655 ]
656 .iter()
657 .any(|p| Path::new(p).exists());
658 checks.push(Check {
659 what: "an sftp server (door a serves the data mount over it)".into(),
660 ok: have,
661 fix: (!have)
662 .then(|| plat.sshd_package().and_then(|p| plat.install(&[p])))
663 .flatten(),
664 });
665 }
666 Door::Nologin => {}
667 }
668
669 checks
670}
671
672fn socket_activated() -> bool {
675 if !cfg!(target_os = "linux") {
676 return false;
677 }
678 std::process::Command::new("systemctl")
679 .args(["is-active", "ssh.socket"])
680 .output()
681 .map(|o| String::from_utf8_lossy(&o.stdout).trim() == "active")
682 .unwrap_or(false)
683}
684
685#[derive(Debug, Clone, PartialEq, Eq)]
692enum InstallAction {
693 Installed,
694 Replaced,
695 AlreadyPresent,
696 Skipped(String),
697}
698
699fn install_authorized_line(
707 cli: &JoinConfigArgs,
708 line: &str,
709 sshd_port: u16,
710) -> Result<InstallAction, String> {
711 if cli.install_key && cli.no_install_key {
712 return Err("--install-key and --no-install-key contradict each other".into());
713 }
714 let wanted = if cli.install_key {
718 true
719 } else if cli.no_install_key {
720 false
721 } else if cli.yes || !prompt::has_tty() {
722 return Ok(InstallAction::Skipped(
723 "installing needs explicit consent: the prompt, or --install-key".to_string(),
724 ));
725 } else {
726 prompt::ask_yn(
727 "install the guardrailed line into this user's \
728 ~/.ssh/authorized_keys now? (only the wizard's own line is \
729 ever touched)",
730 true,
731 )
732 };
733 if !wanted {
734 return Ok(InstallAction::Skipped("declined".to_string()));
735 }
736
737 let ak_path = match cli.authorized_keys.as_deref() {
746 Some(p) => {
747 let p = match p.strip_prefix("~/") {
750 Some(rest) => home_dir().join(rest),
751 None => PathBuf::from(p),
752 };
753 if p.starts_with("/etc/ssh") {
754 return Err(format!(
755 "{} is system sshd configuration — the wizard installs \
756 door keys, never /etc/ssh. Install it there by hand \
757 (the line is in the install notes)",
758 p.display(),
759 ));
760 }
761 p
762 }
763 None => home_dir().join(".ssh").join("authorized_keys"),
764 };
765 let ssh_dir = ak_path
766 .parent()
767 .ok_or("the authorized_keys path has no parent directory")?
768 .to_path_buf();
769 if cli.authorized_keys.is_none() {
775 if !ssh_dir.is_dir() {
776 fs::create_dir_all(&ssh_dir)
777 .map_err(|e| format!("cannot create {}: {e}", ssh_dir.display()))?;
778 set_mode(&ssh_dir, 0o700)?;
779 } else {
780 fix_perms_confirmed(cli, &ssh_dir, 0o700)?;
781 }
782 } else if !ssh_dir.is_dir() {
783 return Err(format!(
784 "{} does not exist — create the directory holding the \
785 authorized_keys file first",
786 ssh_dir.display(),
787 ));
788 }
789 if ak_path.is_symlink() {
793 return Err(format!(
794 "{} is a symlink — install the line manually (it is in the \
795 install notes)",
796 ak_path.display(),
797 ));
798 }
799
800 let content = if ak_path.is_file() {
801 fix_perms_confirmed(cli, &ak_path, 0o600)?;
802 fs::read_to_string(&ak_path)
803 .map_err(|e| format!("cannot read {}: {e}", ak_path.display()))?
804 } else {
805 String::new()
806 };
807
808 let (new_content, outcome) = upsert_authorized_line(&content, line)?;
809 if outcome == UpsertOutcome::Identical {
810 return Ok(InstallAction::AlreadyPresent);
811 }
812 if outcome == UpsertOutcome::Replaced {
813 let confirmed = cli.install_key
814 || (prompt::has_tty()
815 && prompt::ask_yn(
816 "the key is already installed with DIFFERENT options — \
817 replace that line with the wizard's?",
818 true,
819 ));
820 if !confirmed {
821 return Ok(InstallAction::Skipped(
822 "the key is present with different options; left alone".to_string(),
823 ));
824 }
825 }
826
827 let tmp = ssh_dir.join(".authorized_keys.fdl-tmp");
831 fs::write(&tmp, &new_content).map_err(|e| format!("cannot write {}: {e}", tmp.display()))?;
832 set_mode(&tmp, 0o600)?;
833 fs::rename(&tmp, &ak_path)
834 .map_err(|e| format!("cannot move the new authorized_keys into place: {e}"))?;
835
836 if !sshd_listening(sshd_port) {
839 let hint = if cfg!(target_os = "macos") {
840 " (on macOS: System Settings > General > Sharing > Remote Login)"
841 } else {
842 ""
843 };
844 eprintln!(
845 "{}",
846 style::dim(&format!(
847 "fdl join-config: nothing seems to be listening on port \
848 {sshd_port} — the line is installed, but workers cannot \
849 dial until sshd is up{hint}",
850 )),
851 );
852 }
853
854 Ok(match outcome {
855 UpsertOutcome::Appended => InstallAction::Installed,
856 UpsertOutcome::Replaced => InstallAction::Replaced,
857 UpsertOutcome::Identical => unreachable!("handled above"),
858 })
859}
860
861#[derive(Debug, Clone, Copy, PartialEq, Eq)]
862enum UpsertOutcome {
863 Appended,
864 Replaced,
865 Identical,
866}
867
868fn upsert_authorized_line(content: &str, line: &str) -> Result<(String, UpsertOutcome), String> {
873 let wanted =
874 key_material(line).ok_or("the composed authorized_keys line carries no key material")?;
875 let mut out = String::with_capacity(content.len() + line.len() + 2);
876 let mut outcome = UpsertOutcome::Appended;
877 for existing in content.lines() {
878 if key_material(existing) == Some(wanted) {
879 if existing.trim() == line.trim() {
880 return Ok((content.to_string(), UpsertOutcome::Identical));
881 }
882 out.push_str(line);
883 outcome = UpsertOutcome::Replaced;
884 } else {
885 out.push_str(existing);
886 }
887 out.push('\n');
888 }
889 if outcome == UpsertOutcome::Appended {
890 out.push_str(line);
891 out.push('\n');
892 }
893 Ok((out, outcome))
894}
895
896fn key_material(line: &str) -> Option<(&str, &str)> {
901 let mut rest = line.trim();
902 if rest.is_empty() || rest.starts_with('#') {
903 return None;
904 }
905 loop {
906 let mut fields = rest.splitn(2, char::is_whitespace);
907 let first = fields.next()?;
908 let tail = fields.next().unwrap_or("").trim_start();
909 if first.starts_with("ssh-") || first.starts_with("ecdsa-") || first.starts_with("sk-") {
910 let key = tail.split_whitespace().next()?;
911 return Some((first, key));
912 }
913 let mut in_quotes = false;
917 let mut cut = None;
918 for (i, c) in rest.char_indices() {
919 match c {
920 '"' => in_quotes = !in_quotes,
921 c if c.is_whitespace() && !in_quotes => {
922 cut = Some(i);
923 break;
924 }
925 _ => {}
926 }
927 }
928 rest = rest[cut?..].trim_start();
929 }
930}
931
932#[cfg(unix)]
937fn fix_perms_confirmed(cli: &JoinConfigArgs, path: &Path, mode: u32) -> Result<(), String> {
938 use std::os::unix::fs::PermissionsExt;
939 let current = fs::metadata(path)
940 .map_err(|e| format!("cannot stat {}: {e}", path.display()))?
941 .permissions()
942 .mode()
943 & 0o777;
944 if current == mode {
945 return Ok(());
946 }
947 let question = format!(
948 "{} is mode {current:03o}, sshd wants {mode:03o} — apply `chmod \
949 {mode:o} {}`?",
950 path.display(),
951 path.display(),
952 );
953 let apply = cli.install_key || (prompt::has_tty() && prompt::ask_yn(&question, true));
954 if !apply {
955 return Err(format!(
956 "{} stays mode {current:03o} — sshd will refuse the key until \
957 it is {mode:03o}",
958 path.display(),
959 ));
960 }
961 set_mode(path, mode)
962}
963
964#[cfg(not(unix))]
965fn fix_perms_confirmed(_cli: &JoinConfigArgs, _path: &Path, _mode: u32) -> Result<(), String> {
966 Ok(())
967}
968
969#[cfg(unix)]
970fn set_mode(path: &Path, mode: u32) -> Result<(), String> {
971 use std::os::unix::fs::PermissionsExt;
972 fs::set_permissions(path, fs::Permissions::from_mode(mode))
973 .map_err(|e| format!("cannot chmod {}: {e}", path.display()))
974}
975
976#[cfg(not(unix))]
977fn set_mode(_path: &Path, _mode: u32) -> Result<(), String> {
978 Ok(())
979}
980
981fn sshd_listening(port: u16) -> bool {
984 std::net::TcpStream::connect_timeout(
985 &std::net::SocketAddr::from(([127, 0, 0, 1], port)),
986 std::time::Duration::from_millis(300),
987 )
988 .is_ok()
989}
990
991fn resolve_label(cli: &JoinConfigArgs) -> Result<String, String> {
992 if let Some(l) = &cli.label {
993 return Ok(l.clone());
994 }
995 if let Ok(env) = std::env::var("FDL_ENV")
996 && !env.trim().is_empty()
997 {
998 return Ok(env.trim().to_string());
999 }
1000 Err(
1001 "a farm needs a label: `fdl join-config <label>` (or target an \
1002 existing overlay: `fdl @<label> join-config`)"
1003 .to_string(),
1004 )
1005}
1006
1007fn validate_label(label: &str) -> Result<(), String> {
1010 let ok = !label.is_empty()
1011 && label
1012 .chars()
1013 .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_');
1014 if ok {
1015 Ok(())
1016 } else {
1017 Err(format!(
1018 "farm label `{label}` — letters, digits, `-` and `_` only (it \
1019 names fdl.<label>.yml and .fdl/<label>/)"
1020 ))
1021 }
1022}
1023
1024fn confirm(cli: &JoinConfigArgs, question: &str) -> Result<bool, String> {
1028 if cli.yes {
1029 return Ok(true);
1030 }
1031 if !prompt::has_tty() {
1032 return Err(format!(
1033 "non-interactive run needs a decision: {question} \
1034 (pass --yes to accept, or run in a terminal)"
1035 ));
1036 }
1037 Ok(prompt::ask_yn(question, true))
1038}
1039
1040fn ensure_key(cli: &JoinConfigArgs, key_path: &Path, label: &str) -> Result<KeyAction, String> {
1041 let exists = key_path.is_file() && key_path.with_extension("pub").is_file();
1042 if exists {
1043 let regen = if cli.regen {
1046 true
1047 } else if cli.yes || !prompt::has_tty() {
1048 false
1049 } else {
1050 prompt::ask_yn(
1051 &format!(
1052 "a join key already exists for `{label}` — regenerate it \
1053 for a new farm instantiation? (workers holding the old \
1054 key stop being admitted)"
1055 ),
1056 false,
1057 )
1058 };
1059 if !regen {
1060 return Ok(KeyAction::Reused);
1061 }
1062 fs::remove_file(key_path).ok();
1063 fs::remove_file(key_path.with_extension("pub")).ok();
1064 generate_key(key_path, label)?;
1065 return Ok(KeyAction::Regenerated);
1066 }
1067 generate_key(key_path, label)?;
1068 Ok(KeyAction::Generated)
1069}
1070
1071fn generate_key(key_path: &Path, label: &str) -> Result<(), String> {
1072 let out = Command::new("ssh-keygen")
1073 .args(["-t", "ed25519", "-N", "", "-C"])
1074 .arg(format!("flodl-join-{label}"))
1075 .arg("-f")
1076 .arg(key_path)
1077 .output()
1078 .map_err(|e| {
1079 format!(
1080 "cannot run ssh-keygen ({e}) — it ships with OpenSSH, which \
1081 the join sshd needs anyway"
1082 )
1083 })?;
1084 if !out.status.success() {
1085 return Err(format!(
1086 "ssh-keygen failed: {}",
1087 String::from_utf8_lossy(&out.stderr).trim(),
1088 ));
1089 }
1090 #[cfg(unix)]
1093 {
1094 use std::os::unix::fs::PermissionsExt;
1095 let _ = fs::set_permissions(key_path, fs::Permissions::from_mode(0o600));
1096 }
1097 Ok(())
1098}
1099
1100fn ensure_overlay(
1103 cli: &JoinConfigArgs,
1104 overlay_path: &Path,
1105 label: &str,
1106 root: &Path,
1107 cmd_hint: Option<&str>,
1108) -> Result<(String, OverlayAction), String> {
1109 if !overlay_path.is_file() {
1110 let token = fresh_token()?;
1111 let scaffold = render_overlay_scaffold(label, &token, root, cmd_hint);
1112 fs::write(overlay_path, scaffold)
1113 .map_err(|e| format!("cannot write {}: {e}", overlay_path.display()))?;
1114 return Ok((token, OverlayAction::Scaffolded));
1115 }
1116 let content = fs::read_to_string(overlay_path)
1117 .map_err(|e| format!("cannot read {}: {e}", overlay_path.display()))?;
1118 match find_token_line(&content) {
1119 Some(old) => {
1120 let regen = if cli.regen {
1121 true
1122 } else if cli.yes || !prompt::has_tty() {
1123 false
1124 } else {
1125 prompt::ask_yn(
1126 &format!(
1127 "`{label}` already carries a token — mint a fresh one \
1128 for a new farm instantiation? (workers holding the \
1129 old one stop being admitted)"
1130 ),
1131 false,
1132 )
1133 };
1134 if !regen {
1135 return Ok((old, OverlayAction::TokenReused));
1136 }
1137 let token = fresh_token()?;
1138 let replaced = replace_token_line(&content, &token)
1139 .ok_or("token line vanished between read and replace")?;
1140 fs::write(overlay_path, replaced)
1141 .map_err(|e| format!("cannot write {}: {e}", overlay_path.display()))?;
1142 Ok((token, OverlayAction::TokenReplaced))
1143 }
1144 None => {
1145 Ok((fresh_token()?, OverlayAction::SnippetPrinted))
1150 }
1151 }
1152}
1153
1154fn fresh_token() -> Result<String, String> {
1160 let mut bytes = [0u8; 16];
1161 getrandom::fill(&mut bytes)
1162 .map_err(|e| format!("cannot draw OS entropy for the token: {e}"))?;
1163 Ok(bytes.iter().map(|b| format!("{b:02x}")).collect())
1164}
1165
1166fn find_token_line(content: &str) -> Option<String> {
1169 for line in content.lines() {
1170 let t = line.trim_start();
1171 if t.starts_with('#') {
1172 continue;
1173 }
1174 if let Some(rest) = t.strip_prefix("token:") {
1175 let v = rest.trim().trim_matches('"').trim_matches('\'');
1176 if !v.is_empty() {
1177 return Some(v.to_string());
1178 }
1179 }
1180 }
1181 None
1182}
1183
1184fn replace_token_line(content: &str, new_token: &str) -> Option<String> {
1187 let mut out = String::with_capacity(content.len());
1188 let mut replaced = false;
1189 for line in content.lines() {
1190 let t = line.trim_start();
1191 if !replaced
1192 && !t.starts_with('#')
1193 && t.strip_prefix("token:")
1194 .is_some_and(|r| !r.trim().is_empty())
1195 {
1196 let indent = &line[..line.len() - t.len()];
1197 out.push_str(indent);
1198 out.push_str("token: ");
1199 out.push_str(new_token);
1200 replaced = true;
1201 } else {
1202 out.push_str(line);
1203 }
1204 out.push('\n');
1205 }
1206 replaced.then_some(out)
1207}
1208
1209fn recover_shape(farm_dir: &Path) -> Option<(Door, String)> {
1227 let yml = fs::read_to_string(farm_dir.join("worker.yml")).ok()?;
1228 let door = if yml.contains("from: rsync://") {
1229 Door::B
1230 } else if yml.contains("data_source: sshfs://") {
1231 Door::A
1232 } else {
1233 Door::Nologin
1234 };
1235 let field = |key: &str| -> Option<String> {
1236 yml.lines()
1237 .map(str::trim)
1238 .find_map(|l| l.strip_prefix(key))
1239 .map(|v| v.trim().to_string())
1240 };
1241 let host = field("target:")?;
1244 let user = field("user:")?;
1245 let spec = match field("port:") {
1246 Some(p) => format!("{user}@{host}:{p}"),
1247 None => format!("{user}@{host}"),
1248 };
1249 Some((door, spec))
1250}
1251
1252fn command_hint(crate_dir: PathBuf) -> Option<String> {
1257 let manifest = fs::read_to_string(crate_dir.join("Cargo.toml")).ok()?;
1258 package_name(&manifest)
1259}
1260
1261fn render_overlay_scaffold(
1262 label: &str,
1263 token: &str,
1264 root: &Path,
1265 cmd_hint: Option<&str>,
1266) -> String {
1267 let cmd = match cmd_hint {
1271 Some(name) => format!("commands:\n\x20 {name}:\n\x20 cluster: true\n"),
1272 None => {
1273 "# commands:\n# <your-run-command>:\n# cluster: true\n".to_string()
1279 }
1280 };
1281 format!(
1282 "# fdl.{label}.yml — farm overlay, generated by `fdl join-config`.\n\
1283 # Activate with `fdl @{label} <cmd>`. Regenerate credentials with\n\
1284 # `fdl @{label} join-config --regen`. Keys + worker yml live in\n\
1285 # .fdl/{label}/. Deep-merges onto fdl.yml; see\n\
1286 # fdl.cluster-join.yml.example for per-key docs, and `inherit-from:`\n\
1287 # for sharing a base between farms.\n\
1288 \n\
1289 cluster:\n\
1290 \x20 controller:\n\
1291 \x20 host: 127.0.0.1 # loopback; tunnel_only forces it anyway\n\
1292 \x20 port: 1337\n\
1293 \x20 path: {root}\n\
1294 \x20 join:\n\
1295 \x20 discovery: true # the window defines the world\n\
1296 \x20 min_rank_start: 1 # RAISE to your fleet's quorum (in ranks)\n\
1297 \x20 start: manual # hold at quorum; `fdl start` fires it\n\
1298 \x20 join_timeout: 600\n\
1299 \x20 max_join_timeout: 1200\n\
1300 \x20 tunnel_only: true # sshd forward is the only road in (CPU modes)\n\
1301 \x20 token: {token}\n\
1302 \n\
1303 \x20 workers: [] # walk-ins fill it\n\
1304 \n\
1305 # A join window only opens for a command that runs in launcher\n\
1306 # mode, and `cluster:` is what puts it there. Without an entry\n\
1307 # here `fdl @{label} <cmd>` resolves the base command and runs it\n\
1308 # LOCALLY: no window, no walk-ins, and nothing says so, because\n\
1309 # training on this box is a legitimate thing to do. Name the\n\
1310 # command that starts your run.\n\
1311 {cmd}",
1312 label = label,
1313 token = token,
1314 root = root.display(),
1315 cmd = cmd,
1316 )
1317}
1318
1319fn foreign_identity_warning(root: &Path, label: &str, farm_dir: &Path) -> Option<String> {
1323 let base = crate::config::find_project_config(root)?;
1324 let project = crate::config::load_project_with_env(&base, Some(label)).ok()?;
1325 let identity = project.join.as_ref()?.ssh.as_ref()?.identity_file.clone()?;
1326 let p = PathBuf::from(&identity);
1327 if p.starts_with(farm_dir) || identity.contains(&format!(".fdl/{label}/")) {
1328 return None;
1329 }
1330 Some(format!(
1331 "the merged config's join.ssh.identity_file ({identity}) lives \
1332 outside .fdl/{label}/ — a key reused across farms (or anything \
1333 else) widens what one leaked worker can reach. Per-farm keys are \
1334 the point of this wizard.",
1335 ))
1336}
1337
1338fn authorized_keys_line(
1339 door: Door,
1340 served_abs: &Path,
1341 cli: &JoinConfigArgs,
1342 pub_line: &str,
1343) -> String {
1344 let forced = match door {
1345 Door::B => format!("command=\"rrsync -ro {}\"", served_abs.display()),
1346 Door::A => format!(
1347 "command=\"internal-sftp -R -d {}\"",
1348 cli.data_path
1349 .as_deref()
1350 .unwrap_or(crate::config::DEFAULT_DATA_PATH),
1351 ),
1352 Door::Nologin => "command=\"/usr/sbin/nologin\"".to_string(),
1353 };
1354 format!("restrict,port-forwarding,permitopen=\"127.0.0.1:1337\",{forced} {pub_line}")
1355}
1356
1357fn render_sshd_conf(label: &str, door: Door, port: u16, plat: platform::Platform) -> String {
1370 let door_note = match door {
1371 Door::Nologin => "tunnel only",
1372 Door::A => "tunnel + read-only sftp data mount (the key carries the command)",
1373 Door::B => "tunnel + rrsync source pull (the key carries the command)",
1374 };
1375 let mut l: Vec<String> = vec![
1376 format!("# floDl join door for farm `{label}` — generated by `fdl join-config`."),
1377 format!("# Door: {door_note}."),
1378 "#".into(),
1379 format!("# Port {port} is the join door and the only one to expose; 22 stays"),
1380 "# for ordinary logins and should NOT be forwarded from a router.".into(),
1381 ];
1382 if plat == platform::Platform::Debian && port != 22 {
1383 l.push("#".into());
1384 l.push("# NOTE: while ssh.socket owns the listener, the Port line below is".into());
1385 l.push("# IGNORED. The install step disables it and enables ssh.service.".into());
1386 }
1387 if plat == platform::Platform::Rhel && port != 22 {
1388 l.push("#".into());
1389 l.push("# NOTE: SELinux must be told this port is for ssh, or the daemon".into());
1390 l.push("# fails to bind with an error that never mentions SELinux.".into());
1391 }
1392 l.push(String::new());
1393 l.push("Port 22".into());
1394 l.push(format!("Port {port}"));
1395 l.push("PermitRootLogin no".into());
1396 l.push("PasswordAuthentication no".into());
1397 l.push("KbdInteractiveAuthentication no".into());
1398 l.push("PubkeyAuthentication yes".into());
1399 l.push(String::new());
1400 l.push("# The daemon half of the guardrail, bound to the exposed port so".into());
1401 l.push("# ordinary logins on 22 are untouched and ANY key arriving here is".into());
1402 l.push("# confined to the controller mux forward.".into());
1403 l.push(format!("Match LocalPort {port}"));
1404 l.push(" AllowTcpForwarding local".into());
1405 l.push(" PermitOpen 127.0.0.1:1337".into());
1406 if door == Door::Nologin {
1411 l.push(" ForceCommand /usr/sbin/nologin".into());
1412 }
1413 l.push(" PermitTTY no".into());
1414 l.push(" X11Forwarding no".into());
1415 l.push(" AllowAgentForwarding no".into());
1416 l.push(String::new());
1417 l.join("\n")
1418}
1419
1420fn sshd_match_block(user: &str) -> String {
1421 format!(
1422 "Match User {user}\n \
1423 AllowTcpForwarding local\n \
1424 PermitOpen 127.0.0.1:1337\n \
1425 PermitTTY no\n \
1426 X11Forwarding no\n \
1427 AllowAgentForwarding no"
1428 )
1429}
1430
1431struct PublishDerivation {
1435 from_root: PathBuf,
1436 cwd_rel: Option<String>,
1437 bin: String,
1438 build: String,
1439 bin_caveat: Option<String>,
1440}
1441
1442fn derive_publish(crate_dir: &Path) -> Result<Option<PublishDerivation>, String> {
1446 let manifest_path = crate_dir.join("Cargo.toml");
1447 if !manifest_path.is_file() {
1448 return Ok(None);
1449 }
1450 let manifest = fs::read_to_string(&manifest_path)
1451 .map_err(|e| format!("cannot read {}: {e}", manifest_path.display()))?;
1452 let name = package_name(&manifest).ok_or_else(|| {
1453 format!(
1454 "{} has no [package] name — a workspace root? Point --crate at \
1455 the training crate itself",
1456 manifest_path.display()
1457 )
1458 })?;
1459
1460 let crate_abs = crate_dir
1464 .canonicalize()
1465 .map_err(|e| format!("cannot resolve {}: {e}", crate_dir.display()))?;
1466 let (from_root, cwd_rel) = match flodl_path_dep(&manifest) {
1467 Some(rel) => {
1468 let dep_abs = normalize(&crate_abs.join(&rel));
1469 let root = common_ancestor(&crate_abs, &dep_abs);
1470 let cwd_rel = crate_abs
1471 .strip_prefix(&root)
1472 .ok()
1473 .filter(|p| !p.as_os_str().is_empty())
1474 .map(|p| p.display().to_string());
1475 (root, cwd_rel)
1476 }
1477 None => (crate_abs.clone(), None),
1478 };
1479
1480 let bin = format!("target/release/{name}");
1486 let bin_caveat = workspace_above(&crate_abs, &from_root).map(|ws| {
1487 format!(
1488 "{} declares [workspace] above the crate — if it claims the \
1489 crate as a member, the artifact lands in the WORKSPACE \
1490 target/, so `bin:` must point there (e.g. \
1491 `{}target/release/{name}`)",
1492 ws.join("Cargo.toml").display(),
1493 "../".repeat(
1494 crate_abs
1495 .strip_prefix(&ws)
1496 .map(|p| p.components().count())
1497 .unwrap_or(1),
1498 ),
1499 )
1500 });
1501
1502 let features = declares_gpu_features(&manifest);
1503 let build = if features {
1504 format!("cargo build --release --features \"$FDL_GPU_FEATURE\" --bin {name}")
1505 } else {
1506 format!("cargo build --release --bin {name}")
1507 };
1508
1509 Ok(Some(PublishDerivation {
1510 from_root,
1511 cwd_rel,
1512 bin,
1513 build,
1514 bin_caveat,
1515 }))
1516}
1517
1518fn package_name(manifest: &str) -> Option<String> {
1521 let mut in_package = false;
1522 for line in manifest.lines() {
1523 let t = line.trim();
1524 if t.starts_with('[') {
1525 in_package = t == "[package]";
1526 continue;
1527 }
1528 if in_package && let Some(rest) = t.strip_prefix("name") {
1529 let rest = rest.trim_start();
1530 if let Some(v) = rest.strip_prefix('=') {
1531 return Some(v.trim().trim_matches('"').to_string());
1532 }
1533 }
1534 }
1535 None
1536}
1537
1538fn flodl_path_dep(manifest: &str) -> Option<String> {
1542 let mut in_flodl_table = false;
1543 for line in manifest.lines() {
1544 let t = line.trim();
1545 if t.starts_with('[') {
1546 in_flodl_table = t == "[dependencies.flodl]"
1547 || t == "[dev-dependencies.flodl]"
1548 || t == "[dependencies.flodl-hf]";
1549 continue;
1550 }
1551 let inline = t
1552 .strip_prefix("flodl")
1553 .map(|r| r.trim_start())
1554 .and_then(|r| r.strip_prefix('='))
1555 .map(|r| r.trim());
1556 if let Some(spec) = inline {
1557 if let Some(p) = extract_path_value(spec) {
1558 return Some(p);
1559 }
1560 continue;
1561 }
1562 if in_flodl_table
1563 && let Some(rest) = t.strip_prefix("path")
1564 && let Some(v) = rest.trim_start().strip_prefix('=')
1565 {
1566 return Some(v.trim().trim_matches('"').to_string());
1567 }
1568 }
1569 None
1570}
1571
1572fn extract_path_value(spec: &str) -> Option<String> {
1574 let idx = spec.find("path")?;
1575 let rest = spec[idx + 4..].trim_start().strip_prefix('=')?;
1576 let rest = rest.trim_start();
1577 let quoted = rest.strip_prefix('"')?;
1578 let end = quoted.find('"')?;
1579 Some(quoted[..end].to_string())
1580}
1581
1582fn declares_gpu_features(manifest: &str) -> bool {
1585 let mut in_features = false;
1586 for line in manifest.lines() {
1587 let t = line.trim();
1588 if t.starts_with('[') {
1589 in_features = t == "[features]";
1590 continue;
1591 }
1592 if in_features {
1593 let key = t.split('=').next().unwrap_or("").trim();
1594 if key == "cuda" || key == "rocm" {
1595 return true;
1596 }
1597 }
1598 }
1599 false
1600}
1601
1602fn normalize(p: &Path) -> PathBuf {
1605 let mut out = PathBuf::new();
1606 for c in p.components() {
1607 match c {
1608 std::path::Component::ParentDir => {
1609 out.pop();
1610 }
1611 std::path::Component::CurDir => {}
1612 other => out.push(other),
1613 }
1614 }
1615 out
1616}
1617
1618fn common_ancestor(a: &Path, b: &Path) -> PathBuf {
1619 let mut out = PathBuf::new();
1620 for (ca, cb) in a.components().zip(b.components()) {
1621 if ca == cb {
1622 out.push(ca);
1623 } else {
1624 break;
1625 }
1626 }
1627 out
1628}
1629
1630fn workspace_above(crate_abs: &Path, from_root: &Path) -> Option<PathBuf> {
1633 let mut dir = crate_abs.parent()?;
1634 loop {
1635 let m = dir.join("Cargo.toml");
1636 if m.is_file()
1637 && let Ok(content) = fs::read_to_string(&m)
1638 && content.lines().any(|l| l.trim() == "[workspace]")
1639 {
1640 return Some(dir.to_path_buf());
1641 }
1642 if dir == from_root {
1643 return None;
1644 }
1645 dir = dir.parent()?;
1646 }
1647}
1648
1649fn render_publish_block(d: &PublishDerivation) -> String {
1650 let mut out = String::from("publish:\n");
1651 out.push_str(&format!(" source: file://{}\n", d.from_root.display()));
1652 if let Some(cwd) = &d.cwd_rel {
1653 out.push_str(&format!(" cwd: {cwd}\n"));
1654 }
1655 out.push_str(&format!(" build: {}\n", d.build));
1656 out.push_str(&format!(" bin: {}\n", d.bin));
1657 out.push_str(" # args: [--model, ..., --epochs, ...] # the RUN's args\n");
1658 out
1659}
1660
1661fn freshness_report(from_root: &Path) -> String {
1664 let lock = from_root.join("Cargo.lock");
1665 let Ok(lock_meta) = fs::metadata(&lock) else {
1666 return format!(
1667 "no Cargo.lock at {} yet — the publish gate build will create \
1668 the verified pin",
1669 from_root.display(),
1670 );
1671 };
1672 let lock_mtime = lock_meta.modified().ok();
1673 let newest = newest_source_mtime(from_root);
1674 match (lock_mtime, newest) {
1675 (Some(l), Some((n, path))) if n > l => format!(
1676 "Cargo.lock predates the newest source edit ({}) — the next \
1677 gate build refreshes the pin; publish before pointing workers \
1678 at this tree",
1679 path.display(),
1680 ),
1681 (Some(_), Some(_)) => "Cargo.lock is current with the source".to_string(),
1682 _ => "freshness undetermined (no readable source mtimes)".to_string(),
1683 }
1684}
1685
1686fn newest_source_mtime(root: &Path) -> Option<(std::time::SystemTime, PathBuf)> {
1687 let mut newest: Option<(std::time::SystemTime, PathBuf)> = None;
1688 let mut stack = vec![root.to_path_buf()];
1689 while let Some(dir) = stack.pop() {
1690 let Ok(entries) = fs::read_dir(&dir) else {
1691 continue;
1692 };
1693 for entry in entries.flatten() {
1694 let path = entry.path();
1695 let name = entry.file_name();
1696 let name = name.to_string_lossy();
1697 if path.is_dir() {
1698 if matches!(name.as_ref(), "target" | ".git" | ".fdl" | "libtorch") {
1699 continue;
1700 }
1701 stack.push(path);
1702 } else if name != "Cargo.lock"
1703 && let Ok(m) = entry.metadata()
1704 && let Ok(t) = m.modified()
1705 && newest.as_ref().is_none_or(|(n, _)| t > *n)
1706 {
1707 newest = Some((t, path));
1708 }
1709 }
1710 }
1711 newest
1712}
1713
1714fn render_worker_yml(
1718 label: &str,
1719 controller: &Endpoint,
1720 token: &str,
1721 door: Door,
1722 cli: &JoinConfigArgs,
1723) -> String {
1724 let mut out = format!(
1725 "# fdl.yml for a `{label}` farm worker — generated by `fdl join-config`.\n\
1726 # Land the private key at {WORKER_KEY_PATH} (0600) and run:\n\
1727 # fdl join\n\
1728 # (persist: true makes exits re-dial; the systemd recipe in\n\
1729 # fdl.yml.example turns exit 2 into self-deprovisioning.)\n\
1730 \n\
1731 join:\n\
1732 \x20 controller: 127.0.0.1:1337 # the tunnel's loopback end\n\
1733 \x20 ssh:\n\
1734 \x20 target: {host}\n",
1735 label = label,
1736 host = controller.host,
1737 );
1738 if controller.port != 22 {
1739 out.push_str(&format!(" port: {}\n", controller.port));
1740 }
1741 out.push_str(&format!(
1742 " user: {user}\n\
1743 \x20 identity_file: {WORKER_KEY_PATH}\n\
1744 \x20 token: {token}\n\
1745 \x20 libtorch: auto # routes on THIS box's devices\n",
1746 user = controller.user,
1747 token = token,
1748 ));
1749 match door {
1750 Door::B => {
1751 out.push_str(&format!(
1752 " source:\n\
1753 \x20 from: rsync://{user}@{host}:/tree # rrsync re-roots under the served dir\n",
1754 user = controller.user,
1755 host = controller.host,
1756 ));
1757 }
1758 Door::A => {
1759 let data_path = cli
1760 .data_path
1761 .as_deref()
1762 .unwrap_or(crate::config::DEFAULT_DATA_PATH);
1763 out.push_str(&format!(
1764 " data_path: {data_path}\n\
1765 \x20 data_source: sshfs://{user}@{host}:{data_path}\n\
1766 \x20 # door `a` serves the DATA mount; the training binary must be\n\
1767 \x20 # provisioned (`bin:`) or pulled through another road.\n\
1768 \x20 # bin: /path/to/train\n",
1769 user = controller.user,
1770 host = controller.host,
1771 ));
1772 }
1773 Door::Nologin => {
1774 out.push_str(
1775 " # door `nologin` is tunnel-only: provision the binary and any\n\
1776 \x20 # data root, then declare them here.\n\
1777 \x20 # bin: /path/to/train\n\
1778 \x20 # data_path: /flodl/data\n",
1779 );
1780 }
1781 }
1782 if door != Door::A
1783 && let Some(dp) = &cli.data_path
1784 {
1785 out.push_str(&format!(" data_path: {dp}\n"));
1786 }
1787 if let Some(share) = cli.gpu_ram_share {
1788 out.push_str(&format!(
1789 " gpu_ram_share: {share} # this box's APU aperture share\n"
1790 ));
1791 }
1792 out.push_str(" persist: true\n");
1793 out
1794}
1795
1796fn render_notes(
1797 label: &str,
1798 authorized_line: &str,
1799 match_block: &str,
1800 controller: &Endpoint,
1801 door: Door,
1802) -> String {
1803 let door_name = match door {
1804 Door::B => "B (rrsync source pull — publish-then-join)",
1805 Door::A => "A (read-only sftp data mount)",
1806 Door::Nologin => "nologin (tunnel only)",
1807 };
1808 format!(
1809 "# Farm `{label}` — controller-side install notes\n\n\
1810 Door: {door_name}\n\n\
1811 ## 1. authorized_keys ({user}@{host})\n\n\
1812 Append to `~{user}/.ssh/authorized_keys` (one line):\n\n\
1813 ```\n{authorized_line}\n```\n\n\
1814 ## 2. Hardening (optional, recommended for a permanent setup)\n\n\
1815 A dedicated no-shell user plus the daemon-level mirror of the key\n\
1816 restrictions, so a mistake in either layer is caught by the other:\n\n\
1817 ```\n{match_block}\n```\n\n\
1818 ## 3. The worker side\n\n\
1819 Copy `keys/{KEY_NAME}` to each worker at `{WORKER_KEY_PATH}` (0600)\n\
1820 and `worker.yml` to its `fdl.yml`, then `fdl join`. Workers reach\n\
1821 this box at `{host}:{port}`.\n\n\
1822 Full recipe rationale: docs/ddp/02-cluster-guide.md.\n",
1823 label = label,
1824 door_name = door_name,
1825 user = controller.user,
1826 host = controller.host,
1827 port = controller.port,
1828 authorized_line = authorized_line,
1829 match_block = match_block,
1830 )
1831}
1832
1833impl Report {
1834 fn render_human(&self) -> String {
1835 let mut out = String::new();
1836 let push = |out: &mut String, s: &str| {
1837 out.push_str(s);
1838 out.push('\n');
1839 };
1840 push(&mut out, "");
1841 push(&mut out, &format!(" farm: {}", self.label));
1842 push(
1843 &mut out,
1844 &format!(" dir: {}", self.farm_dir.display()),
1845 );
1846 let overlay = match self.overlay_action {
1847 OverlayAction::Scaffolded => "created",
1848 OverlayAction::TokenReplaced => "token regenerated",
1849 OverlayAction::TokenReused => "token reused",
1850 OverlayAction::SnippetPrinted => "NOT edited (user-authored, no token)",
1851 };
1852 push(
1853 &mut out,
1854 &format!(" overlay: {} ({overlay})", self.overlay_path.display()),
1855 );
1856 let key = match self.key_action {
1857 KeyAction::Generated => "generated",
1858 KeyAction::Regenerated => "REGENERATED (old key no longer admits)",
1859 KeyAction::Reused => "reused",
1860 };
1861 push(
1862 &mut out,
1863 &format!(" join key: {} ({key})", self.key_path.display()),
1864 );
1865 push(
1866 &mut out,
1867 &format!(" worker yml: {}", self.worker_yml_path.display()),
1868 );
1869 push(
1870 &mut out,
1871 &format!(" notes: {}", self.notes_path.display()),
1872 );
1873 let install = match &self.install {
1874 InstallAction::Installed => "line appended to ~/.ssh/authorized_keys".to_string(),
1875 InstallAction::Replaced => {
1876 "line REPLACED in ~/.ssh/authorized_keys (options updated)".to_string()
1877 }
1878 InstallAction::AlreadyPresent => "already in ~/.ssh/authorized_keys".to_string(),
1879 InstallAction::Skipped(why) => format!("NOT installed ({why}) — see notes"),
1880 };
1881 push(&mut out, &format!(" sshd: {install}"));
1882 if let Some(ci) = &self.cloud_init_path {
1883 push(
1884 &mut out,
1885 &format!(
1886 " cloud-init: {} (SECRET: key + token inside)",
1887 ci.display()
1888 ),
1889 );
1890 }
1891 if let Some(w) = &self.reuse_warning {
1892 push(&mut out, "");
1893 push(&mut out, &format!(" WARNING: {w}"));
1894 }
1895 if self.overlay_action == OverlayAction::SnippetPrinted {
1896 push(&mut out, "");
1897 push(
1898 &mut out,
1899 " Your overlay carries no token; add under `cluster.controller.join:`:",
1900 );
1901 push(&mut out, "");
1902 push(
1903 &mut out,
1904 " token: <generated — see worker.yml, they must match>",
1905 );
1906 }
1907 push(&mut out, "");
1908 push(
1909 &mut out,
1910 " ── authorized_keys line (controller sshd, one line) ──",
1911 );
1912 push(&mut out, "");
1913 push(&mut out, &format!(" {}", self.authorized_line));
1914 push(&mut out, "");
1915 push(
1916 &mut out,
1917 &format!(
1918 " ── worker fdl.yml ({}) ──",
1919 self.worker_yml_path.display()
1920 ),
1921 );
1922 push(&mut out, "");
1923 for line in self.worker_yml.lines() {
1924 push(&mut out, &format!(" {line}"));
1925 }
1926 if let Some(p) = &self.publish_block {
1927 push(&mut out, "");
1928 push(
1929 &mut out,
1930 " ── publish recipe for the base fdl.yml (then: `fdl publish`) ──",
1931 );
1932 push(&mut out, "");
1933 for line in p.lines() {
1934 push(&mut out, &format!(" {line}"));
1935 }
1936 }
1937 if let Some(c) = &self.bin_caveat {
1938 push(&mut out, "");
1939 push(&mut out, &format!(" note: {c}"));
1940 }
1941 if let Some(f) = &self.freshness {
1942 push(&mut out, "");
1943 push(&mut out, &format!(" freshness: {f}"));
1944 }
1945 push(&mut out, "");
1946 for line in self.steps() {
1947 push(&mut out, &line);
1948 }
1949 push(&mut out, "");
1950 push(
1951 &mut out,
1952 &style::dim(&format!(
1953 " Rationale + hardening notes: {}. The private key is the \
1954 worker-bound secret; it never prints here.",
1955 self.notes_path.display(),
1956 )),
1957 );
1958 push(&mut out, "");
1959 out
1960 }
1961
1962 fn steps(&self) -> Vec<String> {
1967 let mut out = Vec::new();
1968 let mut n = 0;
1969 let mut step = |out: &mut Vec<String>, title: &str, cmds: &[String]| {
1970 n += 1;
1971 out.push(format!(" {n}. {title}"));
1972 for c in cmds {
1973 out.push(format!(" {c}"));
1974 }
1975 out.push(String::new());
1976 };
1977
1978 out.push(format!(
1979 " ── setup, in order ({}{}) ──",
1980 self.plat.name(),
1981 if self.in_container {
1982 ", inside a container"
1983 } else {
1984 ""
1985 },
1986 ));
1987 out.push(String::new());
1988 if self.in_container {
1993 out.push(
1994 " note: this ran INSIDE a container, so the findings \
1995 describe the container."
1996 .to_string(),
1997 );
1998 out.push(
1999 " A package installed into a running container is gone \
2000 after the next `--rm`:"
2001 .to_string(),
2002 );
2003 out.push(
2004 " add it to the image's Dockerfile and rebuild \
2005 (`docker compose build <service>`) to make it stick."
2006 .to_string(),
2007 );
2008 out.push(String::new());
2009 }
2010
2011 let gaps: Vec<&Check> = self.checks.iter().filter(|c| !c.ok).collect();
2014 if !gaps.is_empty() {
2015 let mut cmds: Vec<String> = Vec::new();
2016 for c in &gaps {
2017 cmds.push(format!("# {}", c.what));
2018 match &c.fix {
2019 Some(f) => cmds.push(f.clone()),
2020 None => cmds.push("# (no command for this one here)".to_string()),
2021 }
2022 }
2023 if !self.in_container && !self.docker_services.is_empty() {
2027 cmds.push(String::new());
2028 cmds.push(format!(
2029 "# this project dispatches through docker ({}), so if the \
2030 door or the build lives there, run the above INSIDE it:",
2031 self.docker_services.join(", "),
2032 ));
2033 cmds.push(format!(
2034 "# docker compose exec {} <command>",
2035 self.docker_services
2036 .first()
2037 .map(String::as_str)
2038 .unwrap_or("<service>"),
2039 ));
2040 cmds.push(
2041 "# and add it to that image's Dockerfile, or the next \
2042 `--rm` discards it"
2043 .to_string(),
2044 );
2045 }
2046 step(&mut out, "this box is missing what the door needs:", &cmds);
2047 }
2048
2049 step(
2050 &mut out,
2051 "install the sshd drop-in (read it first — it is yours to edit):",
2052 &{
2053 let mut c = vec![format!(
2054 "sudo install -m 644 {} /etc/ssh/sshd_config.d/flodl-{}.conf",
2055 self.sshd_conf_path.display(),
2056 self.label,
2057 )];
2058 c.extend(self.plat.enable_sshd());
2059 c.push("sudo sshd -t && echo 'sshd config OK'".to_string());
2060 if let Some(fw) = self.plat.open_port(self.controller.port) {
2061 c.push(fw);
2062 }
2063 c
2064 },
2065 );
2066
2067 if !matches!(
2068 self.install,
2069 InstallAction::Installed | InstallAction::Replaced | InstallAction::AlreadyPresent
2070 ) {
2071 step(
2072 &mut out,
2073 "authorize the join key (the wizard can do this for you):",
2074 &[format!("fdl join-config {} --install-key", self.label)],
2075 );
2076 }
2077
2078 let key = self.key_path.display().to_string();
2089 let (p, u, h) = (
2090 self.controller.port,
2091 &self.controller.user,
2092 &self.controller.host,
2093 );
2094 let mut verify = vec![format!(
2095 "ssh -i {key} -p {p} {u}@{h} true # must NOT give a shell"
2096 )];
2097 match self.door {
2098 Door::B => verify.push(format!(
2099 "rsync --list-only -e 'ssh -i {key} -p {p}' {u}@{h}:/tree/ # must LIST the served tree"
2100 )),
2101 Door::A => verify.push(format!(
2102 "sftp -i {key} -P {p} {u}@{h} # must OPEN (read-only)"
2103 )),
2104 Door::Nologin => {}
2105 }
2106 verify.push(format!(
2107 "ssh -i {key} -p {p} {u}@{h} -N -L 19337:127.0.0.1:1337 # must CONNECT"
2108 ));
2109 verify.push(format!(
2110 "sudo sshd -T -C user={u},host=x,addr=127.0.0.1,laddr=127.0.0.1,lport={p} \
2111 | grep -E 'permitopen|forcecommand'"
2112 ));
2113 step(&mut out, "prove the door does exactly one thing:", &verify);
2114
2115 step(
2116 &mut out,
2117 "land the worker's config and key on each box:",
2118 &[
2119 format!(
2120 "scp {} <worker>:{}",
2121 self.key_path.display(),
2122 WORKER_KEY_PATH,
2123 ),
2124 format!("ssh <worker> 'chmod 600 {WORKER_KEY_PATH}'"),
2125 format!(
2126 "scp {} <worker>:<project>/fdl.yml",
2127 self.worker_yml_path.display(),
2128 ),
2129 ],
2130 );
2131
2132 step(
2133 &mut out,
2134 "open the window here, then let the boxes dial in:",
2135 &[
2136 format!(
2137 "fdl @{} <your-run-command> # holds a join window",
2138 self.label
2139 ),
2140 "ssh <worker> 'cd <project> && fdl join'".to_string(),
2141 format!(
2142 "fdl @{} status # then: fdl @{} start",
2143 self.label, self.label
2144 ),
2145 ],
2146 );
2147
2148 out
2149 }
2150
2151 fn to_json(&self) -> serde_json::Value {
2153 serde_json::json!({
2154 "label": self.label,
2155 "farm_dir": self.farm_dir.display().to_string(),
2156 "overlay": {
2157 "path": self.overlay_path.display().to_string(),
2158 "action": match self.overlay_action {
2159 OverlayAction::Scaffolded => "scaffolded",
2160 OverlayAction::TokenReplaced => "token_replaced",
2161 OverlayAction::TokenReused => "token_reused",
2162 OverlayAction::SnippetPrinted => "snippet_printed",
2163 },
2164 },
2165 "key": {
2166 "private_path": self.key_path.display().to_string(),
2167 "public_line": self.pub_line,
2168 "action": match self.key_action {
2169 KeyAction::Generated => "generated",
2170 KeyAction::Regenerated => "regenerated",
2171 KeyAction::Reused => "reused",
2172 },
2173 },
2174 "door": match self.door {
2175 Door::B => "b",
2176 Door::A => "a",
2177 Door::Nologin => "nologin",
2178 },
2179 "authorized_keys_line": self.authorized_line,
2180 "platform": self.plat.name(),
2181 "in_container": self.in_container,
2182 "docker_services": self.docker_services,
2183 "sshd_conf_path": self.sshd_conf_path.display().to_string(),
2184 "preflight": self.checks.iter().map(|c| serde_json::json!({
2185 "what": c.what,
2186 "ok": c.ok,
2187 "fix": c.fix,
2188 })).collect::<Vec<_>>(),
2189 "steps": self.steps(),
2190 "sshd_match_block": self.match_block,
2191 "controller": format!(
2192 "{}@{}:{}",
2193 self.controller.user, self.controller.host, self.controller.port,
2194 ),
2195 "worker_yml_path": self.worker_yml_path.display().to_string(),
2196 "notes_path": self.notes_path.display().to_string(),
2197 "install": match &self.install {
2198 InstallAction::Installed => serde_json::json!({"action": "installed"}),
2199 InstallAction::Replaced => serde_json::json!({"action": "replaced"}),
2200 InstallAction::AlreadyPresent => {
2201 serde_json::json!({"action": "already_present"})
2202 }
2203 InstallAction::Skipped(why) => {
2204 serde_json::json!({"action": "skipped", "why": why})
2205 }
2206 },
2207 "cloud_init_path": self
2208 .cloud_init_path
2209 .as_ref()
2210 .map(|p| p.display().to_string()),
2211 "publish_block": self.publish_block,
2212 "bin_caveat": self.bin_caveat,
2213 "freshness": self.freshness,
2214 "reuse_warning": self.reuse_warning,
2215 })
2216 }
2217}
2218
2219#[cfg(test)]
2220#[path = "join_config_tests.rs"]
2221mod tests;