Runners sign their receipts 47c1d64dec8a
A runner keeps a composite key in its home (runner.key, runner.identity), made on first start. Each receipt is bound to the state of the heads the runner fetched and to the run URL, signed, and sent with finish as signed_receipt. The oxygen pin moves to receipts (Draft 0.8o). closes #10
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-09-29 07:32 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
47c1d64dec8ab70b99464c658a6f7311ae12c5b6db51fdef65d3106b5af63d2d- Depends on
- 635ed0b18ab8 bb643c252515 e0918b887850 fe128fa67a06
- Depended on by
- 0d663e91c598 1215f05ff8d0 3554afefb6aa 5cf1e8aefbb6 4be5e4cd0af6 35f7a849bcc5 ed4aef8b70ca 89171fca079c d60aac6c3f8a b14aeaae683a 6d53fb7d00ec c5b702e59c1f 3007d8f011fa 062c053ef8b1 f98a8f7881e0 a67b97e9a717
- In bookmarks
- main newer-pipeline push-uses-login
- In tags
- v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
- Edits
- 18 atoms, 0 tree operations, 7 files
Browse the tree at this change
modified crates/flow-core/src/api.rs +4 −0
| … | 64 unchanged lines |
||
| 65 | 65 | pub exit_code: Option<i32>, |
|
| 66 | 66 | pub duration_ms: u64, |
|
| 67 | 67 | pub receipt: Option<Receipt>, |
|
| 68 | + | /// The receipt bound to the job's state and signed by the runner |
|
| 69 | + | /// (Oxygen spec §14.6), hex of the signed receipt's bytes. |
|
| 70 | + | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 71 | + | pub signed_receipt: Option<String>, |
|
| 68 | 72 | /// Why the runner could not run the job (rootfs missing, VM failed). |
|
| 69 | 73 | pub error: Option<String>, |
|
| 70 | 74 | } |
|
| … | 5 unchanged lines |
modified crates/flow-core/src/receipt.rs +16 −0
| … | 41 unchanged lines |
||
| 42 | 42 | pub fn hash(&self) -> Hash { |
|
| 43 | 43 | object_hash("receipt", &self.encode()) |
|
| 44 | 44 | } |
|
| 45 | + | ||
| 46 | + | /// The same receipt as an Oxygen object (Oxygen spec §14.6), which a |
|
| 47 | + | /// runner signs and a repository keeps. |
|
| 48 | + | pub fn to_oxy(&self) -> oxy_core::receipt::Receipt { |
|
| 49 | + | use oxy_core::hash::ObjectRef; |
|
| 50 | + | oxy_core::receipt::Receipt { |
|
| 51 | + | job_name: self.job_name.clone(), |
|
| 52 | + | pipeline_hash: ObjectRef(self.pipeline_hash), |
|
| 53 | + | scope_tree_hash: ObjectRef(self.scope_tree_hash), |
|
| 54 | + | inputs: self.inputs.iter().map(|h| ObjectRef(*h)).collect(), |
|
| 55 | + | environment: ObjectRef(self.environment), |
|
| 56 | + | outputs: self.outputs.iter().map(|h| ObjectRef(*h)).collect(), |
|
| 57 | + | exit_code: self.exit_code, |
|
| 58 | + | duration_ms: self.duration_ms, |
|
| 59 | + | } |
|
| 60 | + | } |
|
| 45 | 61 | } |
|
| 46 | 62 | ||
| 47 | 63 | /// What a job ran in (the receipt's `environment`): the pinned pieces of |
|
| … | 53 unchanged lines |
modified crates/flow-core/src/tests.rs +3 −0
| … | 171 unchanged lines |
||
| 172 | 172 | assert_eq!(r.hash(), *blake3::hash(&want).as_bytes()); |
|
| 173 | 173 | let json = serde_json::to_string(&r).unwrap(); |
|
| 174 | 174 | assert_eq!(serde_json::from_str::<Receipt>(&json).unwrap(), r); |
|
| 175 | + | // The same receipt as the Oxygen object a runner signs. |
|
| 176 | + | assert_eq!(r.to_oxy().hash().0, r.hash()); |
|
| 177 | + | assert_eq!(r.to_oxy().encode(), body); |
|
| 175 | 178 | let mut r2 = r.clone(); |
|
| 176 | 179 | r2.exit_code = 1; |
|
| 177 | 180 | assert_ne!(r.hash(), r2.hash()); |
|
| … | 71 unchanged lines |
modified crates/flow/src/home.rs +61 −0
| … | 13 unchanged lines |
||
| 14 | 14 | //! ca = "/path/to/hub-ca.pem" # when the hub's certificate is self-signed |
|
| 15 | 15 | //! name = "earth-1" |
|
| 16 | 16 | //! ``` |
|
| 17 | + | //! |
|
| 18 | + | //! A runner signs its receipts with a composite key kept beside `flow.toml` |
|
| 19 | + | //! (`runner.key`, `runner.identity`), made on first start. |
|
| 17 | 20 | ||
| 18 | 21 | use anyhow::{bail, Context, Result}; |
|
| 19 | 22 | use serde::Deserialize; |
|
| … | 99 unchanged lines |
||
| 119 | 122 | let mut h = blake3::Hasher::new(); |
|
| 120 | 123 | h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?; |
|
| 121 | 124 | Ok(h.finalize().to_hex().to_string()) |
|
| 125 | + | } |
|
| 126 | + | ||
| 127 | + | /// The key a runner signs receipts with (Oxygen spec §14.6). |
|
| 128 | + | pub struct RunnerKey { |
|
| 129 | + | pub identity: oxy_core::identity::Identity, |
|
| 130 | + | pub secret: oxy_core::identity::SecretKey, |
|
| 131 | + | } |
|
| 132 | + | ||
| 133 | + | impl Home { |
|
| 134 | + | /// The runner's key, made on first use as `runner.key` (owner-only) |
|
| 135 | + | /// and `runner.identity`, named `name`. |
|
| 136 | + | pub fn runner_key(&self, name: &str) -> Result<RunnerKey> { |
|
| 137 | + | use oxy_core::identity::{Identity, SecretKey}; |
|
| 138 | + | let key_path = self.root.join("runner.key"); |
|
| 139 | + | let id_path = self.root.join("runner.identity"); |
|
| 140 | + | if key_path.exists() { |
|
| 141 | + | let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?; |
|
| 142 | + | let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?; |
|
| 143 | + | if identity.hash() != key { |
|
| 144 | + | bail!("runner.identity does not belong to runner.key"); |
|
| 145 | + | } |
|
| 146 | + | return Ok(RunnerKey { identity, secret }); |
|
| 147 | + | } |
|
| 148 | + | let secret = SecretKey::generate(); |
|
| 149 | + | let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0); |
|
| 150 | + | let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now); |
|
| 151 | + | std::fs::write(&id_path, identity.encode())?; |
|
| 152 | + | let tmp = self.root.join("runner.key.tmp"); |
|
| 153 | + | { |
|
| 154 | + | use std::io::Write; |
|
| 155 | + | use std::os::unix::fs::OpenOptionsExt; |
|
| 156 | + | let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?; |
|
| 157 | + | f.write_all(&secret.encode(&identity.hash()))?; |
|
| 158 | + | f.sync_all()?; |
|
| 159 | + | } |
|
| 160 | + | std::fs::rename(&tmp, &key_path)?; |
|
| 161 | + | Ok(RunnerKey { identity, secret }) |
|
| 162 | + | } |
|
| 163 | + | } |
|
| 164 | + | ||
| 165 | + | #[cfg(test)] |
|
| 166 | + | mod tests { |
|
| 167 | + | use super::*; |
|
| 168 | + | ||
| 169 | + | #[test] |
|
| 170 | + | fn the_runner_key_is_made_once_and_kept_private() { |
|
| 171 | + | use std::os::unix::fs::PermissionsExt; |
|
| 172 | + | let dir = tempfile::tempdir().unwrap(); |
|
| 173 | + | let home = Home::locate(Some(dir.path())).unwrap(); |
|
| 174 | + | let a = home.runner_key("earth-1").unwrap(); |
|
| 175 | + | assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)"); |
|
| 176 | + | let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode(); |
|
| 177 | + | assert_eq!(mode & 0o777, 0o600); |
|
| 178 | + | let b = home.runner_key("renamed").unwrap(); |
|
| 179 | + | assert_eq!(a.identity, b.identity, "the same key after a restart"); |
|
| 180 | + | let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32])); |
|
| 181 | + | oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap(); |
|
| 182 | + | } |
|
| 122 | 183 | } |
modified crates/flow/src/runner.rs +39 −5
| … | 2 unchanged lines |
||
| 3 | 3 | //! workspace, run it in its tier, stream the log, report the receipt. |
|
| 4 | 4 | //! One job at a time. |
|
| 5 | 5 | ||
| 6 | - | use crate::home::{Home, RunnerConfig}; |
|
| 6 | + | use crate::home::{Home, RunnerConfig, RunnerKey}; |
|
| 7 | 7 | use crate::{microvm, workspace}; |
|
| 8 | 8 | use anyhow::{bail, Context, Result}; |
|
| 9 | 9 | use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, Record, MAX_RECORD_TEXT}; |
|
| … | 90 unchanged lines |
||
| 100 | 100 | let hub = Hub::new(&cfg)?; |
|
| 101 | 101 | let name = cfg.name.clone().unwrap_or_else(|| "flow".into()); |
|
| 102 | 102 | let budget = Budget::of(&cfg); |
|
| 103 | - | eprintln!("flow runner {name}: polling {} (up to {} jobs at once)", cfg.hub, budget.max_jobs); |
|
| 103 | + | let key = home.runner_key(&name)?; |
|
| 104 | + | eprintln!( |
|
| 105 | + | "flow runner {name}: polling {} (up to {} jobs at once), signing receipts as {}", |
|
| 106 | + | cfg.hub, |
|
| 107 | + | budget.max_jobs, |
|
| 108 | + | key.identity.hash().to_hex() |
|
| 109 | + | ); |
|
| 104 | 110 | let used = std::sync::Mutex::new(Used::default()); |
|
| 105 | 111 | let mut idle = Duration::from_secs(2); |
|
| 106 | 112 | std::thread::scope(|scope| loop { |
|
| … | 13 unchanged lines |
||
| 120 | 126 | u.memory_mib += mem; |
|
| 121 | 127 | u.cpus += cpus; |
|
| 122 | 128 | } |
|
| 123 | - | let (home, hub, cfg, used) = (home, &hub, &cfg, &used); |
|
| 129 | + | let (home, hub, cfg, used, key) = (home, &hub, &cfg, &used, &key); |
|
| 124 | 130 | scope.spawn(move || { |
|
| 125 | 131 | let id = claim.job_id; |
|
| 126 | 132 | eprintln!("job {id}: {} {} on {}", claim.repo, claim.job.name, claim.heads.first().map(|h| &h[..12]).unwrap_or("-")); |
|
| 127 | - | let finish = run_claim(home, hub, cfg, &claim).unwrap_or_else(|e| Finish { |
|
| 133 | + | let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(|e| Finish { |
|
| 128 | 134 | exit_code: None, |
|
| 129 | 135 | duration_ms: 0, |
|
| 130 | 136 | receipt: None, |
|
| 137 | + | signed_receipt: None, |
|
| 131 | 138 | error: Some(format!("{e:#}")), |
|
| 132 | 139 | }); |
|
| 133 | 140 | eprintln!( |
|
| … | 56 unchanged lines |
||
| 190 | 197 | /// Held while a job's workspace is prepared (see `run_claim`). |
|
| 191 | 198 | static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(()); |
|
| 192 | 199 | ||
| 193 | - | fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, claim: &Claim) -> Result<Finish> { |
|
| 200 | + | fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> { |
|
| 194 | 201 | let started = Instant::now(); |
|
| 195 | 202 | let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now() }; |
|
| 196 | 203 | let note = |
|
| … | 43 unchanged lines |
||
| 240 | 247 | exit_code: code as u32, |
|
| 241 | 248 | duration_ms: outcome.duration_ms, |
|
| 242 | 249 | }); |
|
| 250 | + | let signed_receipt = match &receipt { |
|
| 251 | + | Some(r) => Some(hex::encode(sign_receipt(&repo, &heads, cfg, key, claim, r)?.encode())), |
|
| 252 | + | None => None, |
|
| 253 | + | }; |
|
| 243 | 254 | Ok(Finish { |
|
| 244 | 255 | exit_code: outcome.exit_code, |
|
| 245 | 256 | duration_ms: started.elapsed().as_millis() as u64, |
|
| 246 | 257 | receipt, |
|
| 258 | + | signed_receipt, |
|
| 247 | 259 | error: outcome.timed_out.then(|| "stopped at the time limit".to_string()), |
|
| 248 | 260 | }) |
|
| 249 | 261 | })(); |
|
| 250 | 262 | let _ = std::fs::remove_dir_all(&scratch); |
|
| 251 | 263 | sink.flush(); |
|
| 252 | 264 | result |
|
| 265 | + | } |
|
| 266 | + | ||
| 267 | + | /// Bind `receipt` to the state of `heads` in `repo` and sign it as the |
|
| 268 | + | /// runner (Oxygen spec §14.6). The state is computed here, not taken from |
|
| 269 | + | /// the claim, so the runner vouches only for what it fetched. |
|
| 270 | + | fn sign_receipt( |
|
| 271 | + | repo: &Repo, |
|
| 272 | + | heads: &[ChangeHash], |
|
| 273 | + | cfg: &RunnerConfig, |
|
| 274 | + | key: &RunnerKey, |
|
| 275 | + | claim: &Claim, |
|
| 276 | + | receipt: &Receipt, |
|
| 277 | + | ) -> Result<oxy_core::receipt::SignedReceipt> { |
|
| 278 | + | let state = state_of(heads); |
|
| 279 | + | let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number); |
|
| 280 | + | Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret)) |
|
| 281 | + | } |
|
| 282 | + | ||
| 283 | + | /// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1). |
|
| 284 | + | pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef { |
|
| 285 | + | use oxy_core::hash::{hash_object, ObjectType}; |
|
| 286 | + | hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode()) |
|
| 253 | 287 | } |
|
| 254 | 288 | ||
| 255 | 289 | fn hex32(s: &str) -> [u8; 32] { |
|
| … | 61 unchanged lines |
replaced deps/oxygen +0 −0
No line changes.
modified docs/m0.md +11 −3
| … | 124 unchanged lines |
||
| 125 | 125 | `(path, content hash)` of the files `scope` matches; the whole tree when |
|
| 126 | 126 | there is no scope), `inputs` (empty in M0), `environment`, `outputs` |
|
| 127 | 127 | (empty), `exit_code`, `duration_ms`. It is encoded as canonical CBOR and |
|
| 128 | - | hashed as `BLAKE3("oxy1 receipt " || len || 0x00 || body)`. M0 stores it on |
|
| 129 | - | the hub with the job; storing receipts as Oxygen objects needs the |
|
| 130 | - | `receipt` object type in the Oxygen spec, and cache reuse (§6.2) follows. |
|
| 128 | + | hashed as `BLAKE3("oxy1 receipt " || len || 0x00 || body)`, the `receipt` |
|
| 129 | + | object of the Oxygen spec (§14.6). |
|
| 130 | + | ||
| 131 | + | **Signed receipts.** A runner keeps a composite key in its home |
|
| 132 | + | (`runner.key`, owner-only, and `runner.identity`), made on first start and |
|
| 133 | + | named after the runner. After each job it binds the receipt to the state of |
|
| 134 | + | the heads it fetched (computed by the runner, not taken from the claim) and |
|
| 135 | + | to the run's URL, signs that attestation, and sends the signed receipt with |
|
| 136 | + | `finish` as `signed_receipt` (hex). The hub checks it and keeps it in the |
|
| 137 | + | repository; anyone can then check it with `oxy receipt verify`. Cache reuse |
|
| 138 | + | (§6.2) follows. |
|
| 131 | 139 | ||
| 132 | 140 | ## 5. Hub side (OxygenHub) |
|
| 133 | 141 | ||
| … | 38 unchanged lines |