Runners ask the hub for their update and take it cedd12342afa

- **Runners ask the hub for their update (oxygenhub#322):** once the hub has answered a claim, and every 15 minutes, a runner asks `GET /api/v1/flow/runner/update`; when the hub offers a build, it downloads it (kept in the home's `updates/`), refuses it unless its size and BLAKE3 are those the hub attests, and installs it as `flow runner update` does, leaving within 600 s, or for a dry run checks it and says what would happen. What it did goes back to the hub once per outcome, and a proof or a rollback is told at the next start. `[runner.update] allow = false` refuses and says so; a hub without runner updates (404) is simply asked again later.

    /// The update policy for this runner and the build it should run

    /// (oxygenhub#322); `None` from a hub without runner updates.

    pub fn update_answer(&self) -> Result<Option<flow_core::api::UpdateAnswer>> {

        let url = format!("{}/api/v1/flow/runner/update", self.base);

        match self.agent.get(&url).set("Authorization", &format!("Bearer {}", self.token)).call() {

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

            Err(ureq::Error::Status(404, _)) => Ok(None),

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

        }

    }

    /// Tell the hub what this runner did about its update.

    pub fn report_update(&self, note: &str) -> Result<()> {

        self.post_with("/flow/runner/update", &flow_core::api::UpdateReport { note: note.to_string() }, &[])?;

        Ok(())

    }

    /// Download an offered build to `to`, refusing it unless its BLAKE3

    /// is the one the hub attests and it is the size it said.

    pub fn download_build(&self, offer: &flow_core::api::UpdateOffer, to: &Path) -> Result<()> {

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

        let url = format!("{}{}", self.base, offer.url);

        let r = self.agent.get(&url).timeout(Duration::from_secs(1800)).call().map_err(|e| anyhow::anyhow!("{url}: {e}"))?;

        let part = to.with_extension("partial");

        let mut out = std::fs::File::create(&part).with_context(|| format!("writing {}", part.display()))?;

        let (mut hasher, mut n, mut buf) = (blake3::Hasher::new(), 0u64, vec![0u8; 1 << 16]);

        let mut body = r.into_reader().take(offer.size + 1);

        loop {

            let k = body.read(&mut buf)?;

            if k == 0 {

                break;

            }

            hasher.update(&buf[..k]);

            out.write_all(&buf[..k])?;

            n += k as u64;

        }

        drop(out);

        let got = hasher.finalize().to_hex().to_string();

        if n != offer.size || got != offer.blake3 {

            let _ = std::fs::remove_file(&part);

            bail!("{} does not match what the hub attests ({n} bytes, BLAKE3 {}…)", offer.asset, &got[..12]);

        }

        std::fs::rename(&part, to)?;

        Ok(())

    }

        // Runner updates (oxygenhub#322): asked of the hub once it has

        // answered a claim, then every 15 minutes.

        let mut update_asked: Option<Instant> = None;

        let mut update_said = String::new();

            }

            if answer.is_ok() && update_asked.is_none_or(|t| t.elapsed() >= UPDATE_EVERY) {

                update_asked = Some(Instant::now());

                if let Some(note) = crate::update::take_note(&home.root) {

                    let _ = hub.report_update(&note);

                }

                consider_update(home, &hub, &mut update_said);

/// How often a runner asks the hub whether it should run another build.

const UPDATE_EVERY: Duration = Duration::from_secs(15 * 60);

/// Notice a runner gives itself to leave for an update the hub offered.

const UPDATE_NOTICE: Duration = Duration::from_secs(600);

/// Ask the hub for this runner's update, and act on it: download the build

/// it offers, check it against the hub's BLAKE3 and as a runner build, and

/// install it, or for a dry run say what would happen. What it did goes to

/// the hub once per outcome (`said`).

fn consider_update(home: &Home, hub: &Hub, said: &mut String) {

    let answer = match hub.update_answer() {

        Ok(Some(a)) => a,

        Ok(None) => return,

        Err(e) => {

            eprintln!("flow runner: asking for an update: {e:#}");

            return;

        }

    };

    let Some(offer) = answer.offer else { return };

    let mut say = |note: String| {

        if *said != note {

            eprintln!("flow runner: {note}");

            if let Err(e) = hub.report_update(&note) {

                eprintln!("flow runner: telling the hub: {e:#}");

            }

            *said = note;

        }

    };

    if home.config.runner.as_ref().and_then(|r| r.update.as_ref()).is_some_and(|u| !u.allow) {

        return say(format!("refuses flow {}: flow.toml says [runner.update] allow = false", offer.version));

    }

    let dir = home.root.join("updates");

    let file = dir.join(format!("flow-{}", offer.version));

    let fresh = std::fs::read(&file).is_ok_and(|b| blake3::hash(&b).to_hex().as_str() == offer.blake3);

    if !fresh {

        if let Err(e) = std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(|_| hub.download_build(&offer, &file)) {

            return say(format!("flow {}: not fetched: {e:#}", offer.version));

        }

    }

    let exe = match std::env::current_exe() {

        Ok(e) => e,

        Err(e) => return say(format!("flow {}: cannot find its own binary: {e}", offer.version)),

    };

    match crate::update::install(home, &exe, &file, UPDATE_NOTICE, answer.policy.dry_run, offer.older) {

        Ok(done) if answer.policy.dry_run => say(format!("dry run: flow {} checks out here and would replace flow {}", done.to, done.from)),

        Ok(done) => {

            let _ = std::fs::remove_file(&file);

            say(format!("installed flow {} (was {}); leaving within {} s to start it", done.to, done.from, UPDATE_NOTICE.as_secs()))

        }

        Err(e) => say(format!("flow {}: not installed: {e:#}", offer.version)),

    }

}

    }

    /// oxygenhub#322: an offered build is kept only if it is the bytes the

    /// hub attests; a peer or a proxy cannot change what is installed.

    #[test]

    fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {

        let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());

        let serve = |bytes: Vec<u8>| {

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

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

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

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

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

                let mut buf = [0u8; 4096];

                let _ = s.read(&mut buf);

                let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());

                let _ = s.write_all(&bytes);

            });

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

        };

        let offer = flow_core::api::UpdateOffer {

            version: "0.3.2".into(),

            asset: "flow-linux-x86_64".into(),

            url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),

            blake3: blake3::hash(&body).to_hex().to_string(),

            size: body.len() as u64,

            ..Default::default()

        };

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

        let to = dir.path().join("flow-0.3.2");

        hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();

        assert_eq!(std::fs::read(&to).unwrap(), body);

        std::fs::remove_file(&to).unwrap();

        let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();

        assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");

        assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");

/// In the runner's home: what to tell the hub once it answers.

const NOTE: &str = "update.note";

/// Keep `note` to tell the hub (a proof, a rollback) once it answers.

fn note(home: &Path, note: &str) {

    let _ = std::fs::write(home.join(NOTE), note);

}

/// What there is to tell the hub, once.

pub fn take_note(home: &Path) -> Option<String> {

    let n = std::fs::read_to_string(home.join(NOTE)).ok()?;

    let _ = std::fs::remove_file(home.join(NOTE));

    Some(n).filter(|n| !n.trim().is_empty())

}

        note(home, &format!("runs flow {} (was {}); proven", p.to, p.from));

    note(home, &format!("rolled back from flow {} to {}: {why}", p.to, p.from));