Register runners with a token; describe them in claims f98a8f7881e0

    /// What the runner is, for the hub's Runners page.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub info: Option<RunnerInfo>,

}

/// A runner's description: shown on the hub's Runners page, and what

/// scheduling by label and architecture will match.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct RunnerInfo {

    /// `x86_64`, `aarch64`, `riscv64`, …

    pub arch: String,

    pub os: String,

    /// The `flow` version.

    pub version: String,

    /// From flow.toml `[runner] labels`.

    #[serde(default)]

    pub labels: Vec<String>,

    /// Jobs at once, and the memory (MiB) and vCPUs they share.

    pub max_jobs: u32,

    #[serde(default)]

    pub memory_mib: Option<u32>,

    #[serde(default)]

    pub cpus: Option<u32>,

}

impl RunnerInfo {

    /// This machine and build, with the given labels and capacity.

    pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> RunnerInfo {

        RunnerInfo {

            arch: std::env::consts::ARCH.into(),

            os: std::env::consts::OS.into(),

            version: env!("CARGO_PKG_VERSION").into(),

            labels,

            max_jobs,

            memory_mib,

            cpus,

        }

    }

}

/// `POST /api/v1/flow/runners/register`: exchange a one-time registration

/// token from the hub for a runner credential, binding the runner's

/// signing identity to it.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Register {

    pub token: String,

    pub name: String,

    /// The runner's Oxygen identity, hex of its encoding: its key signs

    /// every receipt, and the hub accepts no other for this runner.

    pub identity: String,

    pub info: RunnerInfo,

}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Registered {

    pub id: i64,

    pub name: String,

    /// The runner's bearer token from now on; shown once.

    pub credential: String,

/// Where `flow runner register` keeps the runner's hub and credential.

pub const RUNNER_TOML: &str = "runner.toml";

    /// Labels the runner offers, e.g. `gpu`, `big-disk`.

    #[serde(default)]

    pub labels: Vec<String>,

        let mut config: Config = if file.is_file() {

        // `flow runner register` writes the runner's hub and credential to

        // runner.toml, which stands in for a [runner] section flow.toml lacks.

        let registered = root.join(RUNNER_TOML);

        if config.runner.is_none() && registered.is_file() {

            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;

            config.runner = Some(toml::from_str(&text).with_context(|| format!("parse {}", registered.display()))?);

        }

enum RunnerCmd {

    /// Register this runner with a hub using a one-time token from its Admin → Runners page

    Register {

        /// The hub, e.g. https://earth.example:7343

        hub: String,

        /// The registration token (read from stdin when not given)

        #[arg(long)]

        token: Option<String>,

        /// The runner's name (default: the host name)

        #[arg(long)]

        name: Option<String>,

        /// The hub's CA certificate, when self-signed

        #[arg(long)]

        ca: Option<PathBuf>,

        /// A label the runner offers (repeatable)

        #[arg(long)]

        label: Vec<String>,

    },

}

#[derive(Subcommand)]

    },

    /// Claim and run jobs for the hub in flow.toml's [runner] (or the

    /// runner.toml `flow runner register` wrote)

    Runner {

        #[command(subcommand)]

        cmd: Option<RunnerCmd>,

        }

        Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {

            let home = Home::locate(cli.home.as_deref())?;

            if std::env::var_os("OXY_HOME").is_none() {

                std::env::set_var("OXY_HOME", home.root.join("oxy"));

            }

            let name = name.unwrap_or_else(|| {

                std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_else(|_| "flow".into())

            });

            let token = match token {

                Some(t) => t,

                None => {

                    eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");

                    let mut line = String::new();

                    std::io::stdin().read_line(&mut line)?;

                    line.trim().to_string()

                }

            };

            let got = runner::register(&home, &hub, &token, &name, ca, label)?;

            println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);

            Ok(())

        Cmd::Runner { cmd: None } => {

use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};

    }

}

/// `flow runner register`: exchange a one-time registration token from

/// the hub for this runner's credential, binding the runner's signing key,

/// and keep hub, credential and name in runner.toml.

pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {

    if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {

        bail!("flow.toml already has a [runner] section; remove it to register this runner instead");

    }

    let key = home.runner_key(name)?;

    let hub = hub.trim_end_matches('/').to_string();

    let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;

    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();

    let req = Register {

        token: token.trim().to_string(),

        name: name.to_string(),

        identity: hex::encode(key.identity.encode()),

        info: RunnerInfo::here(labels.clone(), 1, None, None),

    };

    let url = format!("{hub}/api/v1/flow/runners/register");

    let got: Registered = match agent.post(&url).send_json(&req) {

        Ok(r) => r.into_json()?,

        Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),

        Err(e) => bail!("{url}: {e}"),

    };

    let mut text = format!(

        "# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",

        got.credential, got.name

    );

    if let Some(ca) = &ca {

        text.push_str(&format!("ca = {:?}\n", ca.display().to_string()));

    }

    if !labels.is_empty() {

        text.push_str(&format!("labels = {labels:?}\n"));

    }

    let path = home.root.join(crate::home::RUNNER_TOML);

    {

        use std::io::Write;

        use std::os::unix::fs::OpenOptionsExt;

        let tmp = path.with_extension("toml.tmp");

        let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;

        f.write_all(text.as_bytes())?;

        f.sync_all()?;

        std::fs::rename(&tmp, &path)?;

    Ok(got)

        let req = ClaimRequest {

            runner: name.clone(),

            runtimes: vec!["microvm".into()],

            free_memory_mib,

            free_cpus,

            info: Some(RunnerInfo::here(cfg.labels.clone(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)),

        };

    }

    /// A hub that answers one request with `body` (status 200) and hands

    /// back what it was sent.

    fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {

        use std::io::{Read, Write};

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();

        let addr = listener.local_addr().unwrap();

        let (tx, rx) = std::sync::mpsc::channel();

        std::thread::spawn(move || {

            let (mut s, _) = listener.accept().unwrap();

            let mut req = Vec::new();

            let mut buf = [0u8; 65536];

            loop {

                let n = s.read(&mut buf).unwrap_or(0);

                if n == 0 {

                    break;

                }

                req.extend_from_slice(&buf[..n]);

                let text = String::from_utf8_lossy(&req);

                if let Some(end) = text.find("\r\n\r\n") {

                    let len = text[..end]

                        .lines()

                        .find_map(|l| {

                            l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))

                        })

                        .unwrap_or(0);

                    if req.len() >= end + 4 + len {

                        break;

                    }

                }

            }

            tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();

            let _ = write!(

                s,

                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",

                body.len()

            );

        });

        (format!("http://{addr}"), rx)

    }

    #[test]

    fn registering_binds_the_key_and_keeps_the_credential_private() {

        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();

        let home = Home::locate(Some(dir.path())).unwrap();

        let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);

        let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();

        assert_eq!((got.id, got.name.as_str()), (7, "plex"));

        let req = sent.recv().unwrap();

        assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");

        let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();

        assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));

        let key = home.runner_key("plex").unwrap();

        assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");

        let path = dir.path().join(crate::home::RUNNER_TOML);

        assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);

        // The runner then starts from runner.toml.

        let again = Home::locate(Some(dir.path())).unwrap();

        let cfg = again.config.runner.unwrap();

        assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));

        assert_eq!(cfg.labels, ["gpu"]);

**Registering a runner.**

- **Token:** a hub administrator creates a one-time registration token on Admin → Runners.

- **Registering:** `flow runner register https://hub --token … [--name N] [--ca PATH] [--label L]…` sends the runner's signing identity, architecture, OS, version and labels (`POST /api/v1/flow/runners/register`). It gets back the runner's own credential and writes hub, credential, name, CA and labels to `runner.toml` (owner-only). That file stands in for a `[runner]` section `flow.toml` lacks, and `flow runner` then starts from it.

- **Key:** registration binds the key; the hub accepts receipts from that runner signed by no other.

- **Claims:** every claim carries the runner's description (`info`: architecture, OS, version, labels, capacity) for the Runners page.

- **Old configuration:** runners configured with a `hub.toml` token still work. Their key is bound on their first signed receipt.