diff --git a/core/crates/ployz-core/src/authority/keys.rs b/core/crates/ployz-core/src/authority/keys.rs index c320f76fe7..45d632e5b5 100644 --- a/core/crates/ployz-core/src/authority/keys.rs +++ b/core/crates/ployz-core/src/authority/keys.rs @@ -27,6 +27,9 @@ pub enum Purpose { /// Standing authority over every future push, such as an automatic deploy rule. Only a /// Device signs one: a session grant covers workflows only. Rule, + /// One Volume workflow, which can stop a Service, start one elsewhere and send a Volume's + /// data to another Machine. Only a Device signs one: a session grant covers none. + Volume, } impl Purpose { @@ -37,6 +40,7 @@ impl Purpose { Self::Join => "join", Self::Enforce => "enforce", Self::Rule => "rule", + Self::Volume => "volume", } } } diff --git a/core/crates/ployz-core/src/authority/mod.rs b/core/crates/ployz-core/src/authority/mod.rs index ec6adb2736..52d52a90cb 100644 --- a/core/crates/ployz-core/src/authority/mod.rs +++ b/core/crates/ployz-core/src/authority/mod.rs @@ -20,6 +20,7 @@ pub mod session; pub mod starting_set; pub mod trusted_key; pub mod validity; +pub mod volume; pub use canonical::{CanonicalError, Timestamp}; pub use deploy::{ @@ -42,3 +43,4 @@ pub use trusted_key::{ Passkey, PasskeyAccount, PasskeyApprover, PasskeyAssertion, PasskeyPublicKey, TrustedKeys, TrustedPasskey, }; +pub use volume::{VolumeApproval, VolumeApproved, VolumeFacts, VolumeStep, VolumeWorkflow}; diff --git a/core/crates/ployz-core/src/authority/refusal.rs b/core/crates/ployz-core/src/authority/refusal.rs index c639175b8b..2c47d46bf9 100644 --- a/core/crates/ployz-core/src/authority/refusal.rs +++ b/core/crates/ployz-core/src/authority/refusal.rs @@ -25,7 +25,7 @@ pub enum AuthorityRefusal { Expired, #[error("this Machine holds no Sealing Key, so it cannot tell which Cluster was approved")] NoClusterIdentity, - #[error("operation {operation} already ran this step; a new restart signs a new approval")] + #[error("operation {operation} already ran this step; it needs a new approval")] Replayed { operation: OperationId }, #[error("this GitHub token already minted a Build Grant; a build asks for a new one")] ProofReplayed, diff --git a/core/crates/ployz-core/src/authority/restart.rs b/core/crates/ployz-core/src/authority/restart.rs index 8f3ed55da4..e2a6f4efa4 100644 --- a/core/crates/ployz-core/src/authority/restart.rs +++ b/core/crates/ployz-core/src/authority/restart.rs @@ -144,7 +144,7 @@ impl RestartApproval { pub use super::validity::{CLOCK_SKEW_MILLIS, MAX_LIFETIME_MILLIS}; -fn ordered_machines<'de, D: serde::Deserializer<'de>>( +pub(super) fn ordered_machines<'de, D: serde::Deserializer<'de>>( deserializer: D, ) -> Result, D::Error> { let machines = Vec::::deserialize(deserializer)?; diff --git a/core/crates/ployz-core/src/authority/volume.rs b/core/crates/ployz-core/src/authority/volume.rs new file mode 100644 index 0000000000..ba3ca9fa70 --- /dev/null +++ b/core/crates/ployz-core/src/authority/volume.rs @@ -0,0 +1,330 @@ +//! The approval a person's Device signs for one Volume workflow, the steps each workflow +//! lets each Machine take, and the checks a Machine runs before it admits the first step. +//! +//! A workflow is a graph, not a list of routes: [`VolumeWorkflow::permits`] answers for +//! one Machine and one step. A Machine keeps its own record of the operations it admitted, +//! so a later step of an open operation needs no fresh window and an ended one never runs +//! again. That record is the Machine's, not this module's. + +use serde::{Deserialize, Serialize}; +use ts_rs::TS; + +use super::{ + canonical::Timestamp, + deploy::ApprovedService, + keys::{Approver, Purpose, Signed, SigningPublicKey, Statement}, + proposal::StepDigest, + refusal::AuthorityRefusal, + restart::{SignerKey, ordered_machines}, + trusted_key::TrustedKeys, +}; +use crate::{ + CopyTransition, DockerVolumeName, MachineId, Namespace, OperationId, + ProvisionedVolumeMaximumBytes, SealingPublicKey, SlotRole, SnapshotGuid, +}; + +/// A person approves one workflow on one Volume in one Cluster, under one operation whose +/// window bounds when its first step may start on each Machine. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TS)] +#[serde(deny_unknown_fields)] +pub struct VolumeApproval { + /// The Cluster's Sealing public key, which every Machine holds and Cloud cannot mint. + pub cluster: SealingPublicKey, + pub operation: OperationId, + pub namespace: Namespace, + /// The Docker Volume the workflow acts on. A seed names the Volume it creates. + pub volume: DockerVolumeName, + pub workflow: VolumeWorkflow, + pub issued_at: Timestamp, + pub expires_at: Timestamp, + /// The Device that signed, which the signature must come from. + pub signer: SignerKey, +} + +impl Statement for VolumeApproval { + const PURPOSE: Purpose = Purpose::Volume; +} + +/// What a person approves, with every Machine it reaches by ID. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TS)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum VolumeWorkflow { + /// The Writer moves from `from` to `to`, which starts `service`. The Mirrors in + /// `mirrors` keep following the new Writer. + Move { + from: MachineId, + to: MachineId, + service: ApprovedService, + /// Sorted and distinct. + #[serde(deserialize_with = "ordered_machines")] + mirrors: Vec, + }, + /// `mirror` keeps a copy of the Writer on `writer`, and every later sync of it, until + /// the Mirror is removed. With `replace` it first destroys the copy it holds. + Mirror { + writer: MachineId, + mirror: MachineId, + replace: bool, + }, + /// `mirror` destroys its copy, held in `role`. `last_copy` says no other copy remains + /// beside the Writer's. + MirrorRemoval { + mirror: MachineId, + writer: MachineId, + role: SlotRole, + last_copy: bool, + }, + /// `machine` promotes its copy at `at` with no start. Writes after `loses_after` are + /// lost. + Restore { + machine: MachineId, + at: SnapshotGuid, + loses_after: Timestamp, + }, + /// Each of `machines` bounds the Volume at `maximum_bytes`, below the bound it holds + /// only when `shrink`. + Resize { + /// Sorted and distinct. + #[serde(deserialize_with = "ordered_machines")] + machines: Vec, + maximum_bytes: ProvisionedVolumeMaximumBytes, + shrink: bool, + }, + /// `machine` creates the approval's Volume from `source` at snapshot `at`. + Seed { + machine: MachineId, + source: DockerVolumeName, + at: SnapshotGuid, + }, +} + +/// One step a Machine is asked to take on a Volume, as its request reads. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum VolumeStep { + /// The `SetVolumeCopy` transition the planner chose for the copy this Machine holds. + Copy(CopyTransition), + /// Starting a Container on the Volume, after a Writer want's transition. + Start, + Seed { + source: DockerVolumeName, + at: SnapshotGuid, + }, + Resize { + maximum_bytes: ProvisionedVolumeMaximumBytes, + allow_shrink: bool, + }, +} + +impl VolumeWorkflow { + /// Whether this workflow lets `machine` take `step`. A step outside the graph is + /// refused however valid the signature. + #[must_use] + pub fn permits(&self, machine: &MachineId, step: &VolumeStep) -> bool { + use CopyTransition::{ + MirrorToForget, MirrorToMirror, MirrorToWriter, NoneToMirror, WriterToMirror, + WriterToSnapshot, WriterToWriter, + }; + match (self, step) { + ( + Self::Move { + from, to, mirrors, .. + }, + VolumeStep::Copy(transition), + ) => { + (machine == from + && matches!( + transition, + WriterToSnapshot | WriterToMirror | WriterToWriter + )) + || (machine == to + && matches!(transition, NoneToMirror | MirrorToMirror | MirrorToWriter)) + || (mirrors.contains(machine) && *transition == MirrorToMirror) + } + (Self::Move { from, to, .. }, VolumeStep::Start) => machine == from || machine == to, + ( + Self::Mirror { + writer, + mirror, + replace, + }, + VolumeStep::Copy(transition), + ) => { + (machine == writer && *transition == WriterToSnapshot) + || (machine == mirror + && (matches!(transition, NoneToMirror | MirrorToMirror) + || (*replace && *transition == MirrorToForget))) + } + (Self::MirrorRemoval { mirror, .. }, VolumeStep::Copy(transition)) => { + machine == mirror && *transition == MirrorToForget + } + (Self::Restore { machine: owner, .. }, VolumeStep::Copy(transition)) => { + machine == owner && *transition == MirrorToWriter + } + ( + Self::Resize { + machines, + maximum_bytes, + shrink, + }, + VolumeStep::Resize { + maximum_bytes: wanted, + allow_shrink, + }, + ) => { + machines.contains(machine) && wanted == maximum_bytes && (*shrink || !allow_shrink) + } + ( + Self::Seed { + machine: owner, + source, + at, + }, + VolumeStep::Seed { + source: wanted, + at: wanted_at, + }, + ) => machine == owner && wanted == source && wanted_at == at, + _ => false, + } + } + + /// Whether this workflow drives `machine`'s copy: changes its role, destroys it, or, + /// for a Mirror a Move keeps following, says where it pulls from. A Machine holds at + /// most one open owner of a Volume's copy. + #[must_use] + pub fn owns(&self, machine: &MachineId) -> bool { + match self { + Self::Move { + from, to, mirrors, .. + } => machine == from || machine == to || mirrors.contains(machine), + Self::Mirror { mirror, .. } | Self::MirrorRemoval { mirror, .. } => machine == mirror, + Self::Restore { machine: owner, .. } => machine == owner, + Self::Resize { .. } | Self::Seed { .. } => false, + } + } + + /// Whether this workflow sends the Volume to `recipient` as a Mirror: a Mirror's own, + /// or one a Move keeps following. + #[must_use] + pub fn mirrors_to(&self, recipient: &MachineId) -> bool { + match self { + Self::Mirror { mirror, .. } => mirror == recipient, + Self::Move { mirrors, .. } => mirrors.contains(recipient), + Self::MirrorRemoval { .. } + | Self::Restore { .. } + | Self::Resize { .. } + | Self::Seed { .. } => false, + } + } + + /// Whether `machine` is the source of a Move under this workflow. + #[must_use] + pub fn moves_from(&self, machine: &MachineId) -> bool { + matches!(self, Self::Move { from, .. } if from == machine) + } + + fn names(&self, machine: &MachineId) -> bool { + match self { + Self::Move { + from, to, mirrors, .. + } => machine == from || machine == to || mirrors.contains(machine), + Self::Mirror { writer, mirror, .. } | Self::MirrorRemoval { mirror, writer, .. } => { + machine == writer || machine == mirror + } + Self::Restore { machine: owner, .. } | Self::Seed { machine: owner, .. } => { + machine == owner + } + Self::Resize { machines, .. } => machines.contains(machine), + } + } + + /// A Move to itself, or a Mirror of its own Writer, would let one Machine take both + /// sides of the graph. + fn distinct(&self) -> bool { + match self { + Self::Move { + from, to, mirrors, .. + } => from != to && !mirrors.contains(from) && !mirrors.contains(to), + Self::Mirror { writer, mirror, .. } | Self::MirrorRemoval { mirror, writer, .. } => { + writer != mirror + } + Self::Restore { .. } | Self::Resize { .. } | Self::Seed { .. } => true, + } + } +} + +/// What a Machine knows when it checks a Volume approval against one step. +#[derive(Clone, Debug)] +pub struct VolumeFacts<'a> { + /// The Sealing public key this Machine holds, or `None` while it holds none. + pub cluster: Option, + pub machine: &'a MachineId, + pub trusted: &'a TrustedKeys, + pub volume: &'a DockerVolumeName, + pub step: &'a VolumeStep, + pub now: Timestamp, +} + +/// A Volume approval a Machine checked: its body, the Device that signed it, and the +/// digest of the body as signed, which the Machine's record keeps. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct VolumeApproved { + pub body: VolumeApproval, + pub by: SigningPublicKey, + pub digest: StepDigest, +} + +/// Check `approval` lets this Machine take `facts.step` now, signed by a Device this +/// Machine trusts. A browser key is refused even under a valid session grant, and so is an +/// envelope that rides a proposal. +/// +/// # Errors +/// Returns the first [`AuthorityRefusal`] found: a signature problem, a malformed body, a +/// signer that is not a trusted Device, a body naming another Cluster or Volume, a +/// workflow that does not name this Machine or does not permit the step, an invalid +/// lifetime, or an expired window. An ended operation is the Machine's record to refuse. +pub fn verify( + approval: &Signed, + facts: &VolumeFacts<'_>, +) -> Result { + let approvers = approval.approvers(facts.trusted, facts.now)?; + let body = approval.open()?; + if approval.proposal.is_some() { + return Err(AuthorityRefusal::Mismatch { + field: "a proposal", + }); + } + let by = approvers + .into_iter() + .find_map(|approver| match approver { + Approver::Device(key) if approver.signs_as(body.signer) => Some(key), + Approver::Device(_) | Approver::Browser { .. } => None, + }) + .ok_or(AuthorityRefusal::Mismatch { + field: "the signer", + })?; + let cluster = facts.cluster.ok_or(AuthorityRefusal::NoClusterIdentity)?; + let checks = [ + ("a Cluster", body.cluster == cluster), + ("a Volume", &body.volume == facts.volume), + ("Machines", body.workflow.distinct()), + ("this Server", body.workflow.names(facts.machine)), + ( + "a Volume step", + body.workflow.permits(facts.machine, facts.step), + ), + ]; + if let Some((field, _)) = checks.iter().find(|(_, holds)| !holds) { + return Err(AuthorityRefusal::Mismatch { field }); + } + super::validity::verify(body.issued_at, body.expires_at, facts.now)?; + Ok(VolumeApproved { + body, + by, + digest: StepDigest::of(&approval.body), + }) +} + +#[cfg(test)] +#[path = "volume_tests.rs"] +mod tests; diff --git a/core/crates/ployz-core/src/authority/volume_tests.rs b/core/crates/ployz-core/src/authority/volume_tests.rs new file mode 100644 index 0000000000..24ffed957c --- /dev/null +++ b/core/crates/ployz-core/src/authority/volume_tests.rs @@ -0,0 +1,511 @@ +use std::num::NonZeroU64; + +use super::*; +use crate::{ + DeviceKey, ServiceName, + authority::{ + canonical, + deploy::PayloadDigest, + keys::Signer, + proposal::{ProposalApproval, ProposalStep, StepKind}, + session::{ActionFamily, Delegation, GrantBody, SessionGrant}, + trusted_key::tests::{Answer, Authenticator, HONEST, trusting}, + }, +}; + +const OPERATION: &str = "0f8fad5bd9cb469fa16570867728950e"; + +fn signer() -> Signer { + Signer::from_seed(&[7; 32]) +} + +fn id(fill: char) -> MachineId { + MachineId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn a() -> MachineId { + id('a') +} + +fn b() -> MachineId { + id('b') +} + +fn c() -> MachineId { + id('c') +} + +fn volume() -> DockerVolumeName { + DockerVolumeName::parse("app_data").unwrap() +} + +fn cluster() -> SealingPublicKey { + SealingPublicKey::from_bytes([9; 32]) +} + +fn bytes(gb: u64) -> ProvisionedVolumeMaximumBytes { + ProvisionedVolumeMaximumBytes::new(NonZeroU64::new(gb * 1_000_000_000).unwrap()) +} + +fn at(millis: u64) -> Timestamp { + Timestamp::from_millis(millis).unwrap() +} + +fn moving() -> VolumeWorkflow { + VolumeWorkflow::Move { + from: a(), + to: b(), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::try_from("d".repeat(64)).unwrap(), + image: None, + }, + mirrors: vec![c()], + } +} + +fn workflows() -> [VolumeWorkflow; 6] { + [ + moving(), + VolumeWorkflow::Mirror { + writer: a(), + mirror: b(), + replace: false, + }, + VolumeWorkflow::MirrorRemoval { + mirror: b(), + writer: a(), + role: SlotRole::Mirror, + last_copy: true, + }, + VolumeWorkflow::Restore { + machine: b(), + at: SnapshotGuid::new(18_446_744_073_709_551_615), + loses_after: at(500_000), + }, + VolumeWorkflow::Resize { + machines: vec![a(), b()], + maximum_bytes: bytes(20), + shrink: false, + }, + VolumeWorkflow::Seed { + machine: a(), + source: DockerVolumeName::parse("app_old").unwrap(), + at: SnapshotGuid::new(42), + }, + ] +} + +fn approval(workflow: VolumeWorkflow) -> VolumeApproval { + VolumeApproval { + cluster: cluster(), + operation: OperationId::parse(OPERATION).unwrap(), + namespace: Namespace::parse("app").unwrap(), + volume: volume(), + workflow, + issued_at: at(100_000), + expires_at: at(1_000_000), + signer: SignerKey::Device(DeviceKey::from(signer().public())), + } +} + +fn trusted() -> TrustedKeys { + TrustedKeys { + devices: vec![DeviceKey::from(signer().public())], + passkeys: Vec::new(), + } +} + +fn check( + signed: &Signed, + machine: &MachineId, + step: &VolumeStep, + now: u64, +) -> Result { + verify( + signed, + &VolumeFacts { + cluster: Some(cluster()), + machine, + trusted: &trusted(), + volume: &volume(), + step, + now: at(now), + }, + ) +} + +fn cut() -> VolumeStep { + VolumeStep::Copy(CopyTransition::WriterToMirror) +} + +fn mismatch(field: &'static str) -> Result { + Err(AuthorityRefusal::Mismatch { field }) +} + +#[test] +fn each_workflow_signs_its_golden_canonical_body() { + let head = r#"{"cluster":"CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=","expires_at":1000000,"issued_at":100000,"namespace":"app","operation":"0f8fad5bd9cb469fa16570867728950e","signer":"ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c","volume":"app_data","workflow":"#; + let (a, b, c) = ("a".repeat(32), "b".repeat(32), "c".repeat(32)); + let d = "d".repeat(64); + let golden = [ + format!( + r#"{{"from":"{a}","kind":"move","mirrors":["{c}"],"service":{{"digest":"{d}","name":"db"}},"to":"{b}"}}"# + ), + format!(r#"{{"kind":"mirror","mirror":"{b}","replace":false,"writer":"{a}"}}"#), + format!( + r#"{{"kind":"mirror_removal","last_copy":true,"mirror":"{b}","role":"mirror","writer":"{a}"}}"# + ), + format!( + r#"{{"at":"18446744073709551615","kind":"restore","loses_after":500000,"machine":"{b}"}}"# + ), + format!( + r#"{{"kind":"resize","machines":["{a}","{b}"],"maximum_bytes":20000000000,"shrink":false}}"# + ), + format!(r#"{{"at":"42","kind":"seed","machine":"{a}","source":"app_old"}}"#), + ]; + for (workflow, golden) in workflows().into_iter().zip(golden) { + let signed = Signed::sign(&approval(workflow.clone()), &signer()).unwrap(); + assert_eq!(signed.body, format!("{head}{golden}}}"), "{workflow:?}"); + assert_eq!(signed.open().unwrap(), approval(workflow)); + assert!( + signed + .signing_input() + .starts_with(b"ployz-authority/v1\nvolume\n") + ); + } +} + +#[test] +fn a_device_approval_admits_each_step_its_graph_names_on_each_machine() { + use CopyTransition::*; + let moved = Signed::sign(&approval(moving()), &signer()).unwrap(); + for (machine, step) in [ + (a(), VolumeStep::Copy(WriterToSnapshot)), + (a(), cut()), + (a(), VolumeStep::Copy(WriterToWriter)), + (a(), VolumeStep::Start), + (b(), VolumeStep::Copy(NoneToMirror)), + (b(), VolumeStep::Copy(MirrorToMirror)), + (b(), VolumeStep::Copy(MirrorToWriter)), + (b(), VolumeStep::Start), + (c(), VolumeStep::Copy(MirrorToMirror)), + ] { + let approved = check(&moved, &machine, &step, 200_000).unwrap(); + assert_eq!(approved.by, signer().public()); + assert_eq!(approved.digest, StepDigest::of(&moved.body)); + } + for (machine, step) in [ + (a(), VolumeStep::Copy(MirrorToWriter)), + (a(), VolumeStep::Copy(WriterToForget)), + (b(), cut()), + (b(), VolumeStep::Copy(MirrorToForget)), + (c(), VolumeStep::Copy(MirrorToWriter)), + (c(), VolumeStep::Start), + ( + a(), + VolumeStep::Resize { + maximum_bytes: bytes(20), + allow_shrink: false, + }, + ), + ] { + assert_eq!( + check(&moved, &machine, &step, 200_000), + mismatch("a Volume step"), + "{machine} {step:?}" + ); + } +} + +#[test] +fn the_other_workflows_permit_only_their_own_steps() { + use CopyTransition::*; + let [_, mirror, removal, restore, resize, seed] = workflows(); + let shrink = |allow_shrink| VolumeStep::Resize { + maximum_bytes: bytes(20), + allow_shrink, + }; + let seeding = |source: &str, guid| VolumeStep::Seed { + source: DockerVolumeName::parse(source).unwrap(), + at: SnapshotGuid::new(guid), + }; + for (workflow, machine, step, permitted) in [ + (&mirror, a(), VolumeStep::Copy(WriterToSnapshot), true), + (&mirror, b(), VolumeStep::Copy(NoneToMirror), true), + (&mirror, b(), VolumeStep::Copy(MirrorToMirror), true), + (&mirror, b(), VolumeStep::Copy(MirrorToForget), false), + (&mirror, b(), VolumeStep::Copy(MirrorToWriter), false), + (&mirror, a(), cut(), false), + (&mirror, b(), VolumeStep::Start, false), + (&removal, b(), VolumeStep::Copy(MirrorToForget), true), + (&removal, a(), VolumeStep::Copy(WriterToForget), false), + (&restore, b(), VolumeStep::Copy(MirrorToWriter), true), + (&restore, b(), VolumeStep::Start, false), + (&resize, a(), shrink(false), true), + (&resize, b(), shrink(true), false), + (&seed, a(), seeding("app_old", 42), true), + (&seed, a(), seeding("other", 42), false), + (&seed, a(), seeding("app_old", 43), false), + ] { + assert_eq!( + workflow.permits(&machine, &step), + permitted, + "{workflow:?} {machine} {step:?}" + ); + } + let replacing = VolumeWorkflow::Mirror { + writer: a(), + mirror: b(), + replace: true, + }; + assert!(replacing.permits(&b(), &VolumeStep::Copy(MirrorToForget))); + let resize_other = VolumeStep::Resize { + maximum_bytes: bytes(30), + allow_shrink: false, + }; + assert!(!resize.permits(&a(), &resize_other)); +} + +#[test] +fn each_refusal_names_what_the_approval_lacks() { + let signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let other_volume = DockerVolumeName::parse("app_other").unwrap(); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(cluster()), + machine: &a(), + trusted: &trusted(), + volume: &other_volume, + step: &cut(), + now: at(200_000), + }, + ), + mismatch("a Volume") + ); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(SealingPublicKey::from_bytes([8; 32])), + machine: &a(), + trusted: &trusted(), + volume: &volume(), + step: &cut(), + now: at(200_000), + }, + ), + mismatch("a Cluster") + ); + assert_eq!( + check(&signed, &id('e'), &cut(), 200_000), + mismatch("this Server") + ); + assert_eq!( + check(&signed, &b(), &cut(), 200_000), + mismatch("a Volume step") + ); + + let mut changed = signed.clone(); + changed.body = changed.body.replace(&"b".repeat(32), &"e".repeat(32)); + assert_eq!( + check(&changed, &id('e'), &VolumeStep::Start, 200_000), + Err(AuthorityRefusal::BadSignature { + key: signer().public() + }) + ); + + let stranger = Signer::from_seed(&[3; 32]); + assert_eq!( + check( + &Signed::sign(&approval(moving()), &stranger).unwrap(), + &a(), + &cut(), + 200_000 + ), + Err(AuthorityRefusal::UnknownKey { + key: stranger.public() + }) + ); + + let mut forged = signed.clone(); + let [signature] = forged.signatures.as_mut_slice() else { + panic!("one signature"); + }; + signature.signature[0] ^= 1; + assert_eq!( + check(&forged, &a(), &cut(), 200_000), + Err(AuthorityRefusal::BadSignature { + key: signer().public() + }) + ); + + let mut unsigned = signed.clone(); + unsigned.signatures.clear(); + assert_eq!( + check(&unsigned, &a(), &cut(), 200_000), + Err(AuthorityRefusal::Unsigned { + purpose: Purpose::Volume + }) + ); + + let mut to_itself = approval(moving()); + let VolumeWorkflow::Move { to, .. } = &mut to_itself.workflow else { + unreachable!("a Move") + }; + *to = a(); + assert_eq!( + check( + &Signed::sign(&to_itself, &signer()).unwrap(), + &a(), + &cut(), + 200_000 + ), + mismatch("Machines") + ); +} + +#[test] +fn the_window_bounds_the_first_step_with_clock_skew() { + let signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let skew = crate::authority::validity::CLOCK_SKEW_MILLIS; + assert!(check(&signed, &a(), &cut(), 1_000_000 + skew).is_ok()); + assert_eq!( + check(&signed, &a(), &cut(), 1_000_001 + skew), + Err(AuthorityRefusal::Expired) + ); + let early = VolumeApproval { + issued_at: at(300_000), + ..approval(moving()) + }; + assert_eq!( + check( + &Signed::sign(&early, &signer()).unwrap(), + &a(), + &cut(), + 299_999 - skew + ), + mismatch("an issue time") + ); + let long = VolumeApproval { + expires_at: at(100_000 + crate::authority::validity::MAX_LIFETIME_MILLIS + 1), + ..approval(moving()) + }; + assert_eq!( + check( + &Signed::sign(&long, &signer()).unwrap(), + &a(), + &cut(), + 200_000 + ), + mismatch("a lifetime") + ); +} + +#[test] +fn a_browser_key_under_a_grant_of_every_family_cannot_sign_a_volume_approval() { + let browser = Signer::from_seed(&[4; 32]); + let passkey = Authenticator::es256(4); + let grant = GrantBody { + browser_key: browser.public(), + cluster: cluster(), + environment: Namespace::parse("app").unwrap(), + environment_id: uuid::Uuid::from_u128(1), + machines: vec![a(), b(), c()], + families: vec![ + ActionFamily::Deploy, + ActionFamily::Restart, + ActionFamily::Settings, + ActionFamily::Start, + ActionFamily::Stop, + ], + issued_at: at(0), + expires_at: at(3_000_000), + delegation: Delegation::None, + }; + let body = canonical::to_canonical(&grant).unwrap(); + let assertion = passkey.answer(&Answer { + challenge: &GrantBody::challenge(&body), + ..HONEST + }); + let browser_approval = VolumeApproval { + signer: SignerKey::Browser { + browser: browser.public(), + }, + ..approval(moving()) + }; + let signed = Signed::sign_under( + &browser_approval, + &browser, + SessionGrant { body, assertion }, + ) + .unwrap(); + let trusted = trusting(&passkey); + assert!( + signed.approvers(&trusted, at(200_000)).is_ok(), + "the grant itself verifies" + ); + assert_eq!( + verify( + &signed, + &VolumeFacts { + cluster: Some(cluster()), + machine: &a(), + trusted: &trusted, + volume: &volume(), + step: &cut(), + now: at(200_000), + }, + ), + mismatch("the signer") + ); +} + +#[test] +fn a_signed_approval_riding_a_proposal_is_refused() { + let mut signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + let bundle = ProposalApproval { + cluster: cluster(), + environment: Namespace::parse("app").unwrap(), + operation: OperationId::parse(OPERATION).unwrap(), + steps: vec![ProposalStep { + index: 0, + kind: StepKind::Deploy, + digest: StepDigest::of(&signed.body), + }], + issued_at: at(100_000), + expires_at: at(1_000_000), + signer: SignerKey::Device(DeviceKey::from(signer().public())), + }; + signed.proposal = Some(Box::new(Signed::sign(&bundle, &signer()).unwrap())); + assert_eq!( + check(&signed, &a(), &cut(), 200_000), + mismatch("a proposal") + ); +} + +#[test] +fn a_field_this_machine_cannot_read_is_malformed() { + let mut signed = Signed::sign(&approval(moving()), &signer()).unwrap(); + signed.body = signed + .body + .replace(r#""kind":"move","#, r#""kind":"move","keep":true,"#); + assert!(matches!( + signed.open(), + Err(AuthorityRefusal::Malformed { + purpose: Purpose::Volume, + .. + }) + )); + let mut unsorted = Signed::sign(&approval(workflows()[4].clone()), &signer()).unwrap(); + let (a, b) = ("a".repeat(32), "b".repeat(32)); + unsorted.body = unsorted + .body + .replace(&format!(r#"["{a}","{b}"]"#), &format!(r#"["{b}","{a}"]"#)); + assert!(matches!( + unsorted.open(), + Err(AuthorityRefusal::Malformed { .. }) + )); +} diff --git a/core/crates/ployz-sdk/authority.d.ts b/core/crates/ployz-sdk/authority.d.ts index 6dfe0908e9..d13ade1aae 100644 --- a/core/crates/ployz-sdk/authority.d.ts +++ b/core/crates/ployz-sdk/authority.d.ts @@ -18,6 +18,8 @@ export type { SignerKey, StepDigest, StepKind, + VolumeApproval, + VolumeWorkflow, } from "./generated/payloads"; import type { ApprovedChange, DeployTerms, PlanView } from "./generated/payloads"; diff --git a/core/crates/ployz/src/sdk/payloads.rs b/core/crates/ployz/src/sdk/payloads.rs index d146787e2d..d82f2c0ba1 100644 --- a/core/crates/ployz/src/sdk/payloads.rs +++ b/core/crates/ployz/src/sdk/payloads.rs @@ -93,6 +93,7 @@ fn typescript_declarations() -> String { declarations.add::(); declarations.add::(); declarations.add::(); + declarations.add::(); declarations.add::(); declarations.add::(); declarations.add::(); diff --git a/core/crates/ployzd/src/machine/local_machine.rs b/core/crates/ployzd/src/machine/local_machine.rs index 863e3b45bd..a6dd08ece6 100644 --- a/core/crates/ployzd/src/machine/local_machine.rs +++ b/core/crates/ployzd/src/machine/local_machine.rs @@ -216,6 +216,9 @@ pub enum Error { /// or its `SeedVolume`. The refusal names `capability` in its details. #[error("{}", volume_approval::refusal_text(.capability))] VolumeRefused { capability: &'static str }, + /// A Volume step that no Volume approval admits on this Machine. + #[error("This Server refused the Volume step: {0}")] + VolumeStepRefused(ployz_core::authority::AuthorityRefusal), #[error("ployz.machine.join.device-approval.v1 needs a Device-signed join approval: {0}")] JoinRefused(ployz_core::authority::AuthorityRefusal), #[error("{SET_DEVICE_ENFORCEMENT_CAPABILITY} needs a Device-signed starting set: {0}")] diff --git a/core/crates/ployzd/src/machine/local_machine/restart_approval.rs b/core/crates/ployzd/src/machine/local_machine/restart_approval.rs index 41174894f5..a62901e29c 100644 --- a/core/crates/ployzd/src/machine/local_machine/restart_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/restart_approval.rs @@ -22,7 +22,7 @@ use ployz_core::{ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use super::{Error, LocalMachine}; +use super::{Error, LocalMachine, volume_approval::VolumeRecords}; const FILE_NAME: &str = "restart-approvals.json"; @@ -242,6 +242,9 @@ struct Consumption { /// The hook each rule's Service last ran here, one per push. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] rule_hooks: BTreeMap, + /// Volume operations, grants and withdrawals, none ever pruned. + #[serde(default, skip_serializing_if = "VolumeRecords::is_empty")] + volumes: VolumeRecords, #[serde(flatten)] operations: BTreeMap, } @@ -856,6 +859,37 @@ impl ConsumptionLog { Ok(guard) } + /// Apply `change` to this Machine's Volume records at `now`, raised to the log's time + /// floor, and keep the result, synced, before returning its answer. A refusal keeps the + /// records as they were, and an unchanged record is not written. + pub(super) fn change_volumes( + &self, + now: Timestamp, + change: impl FnOnce(&mut VolumeRecords, Timestamp) -> Result, + ) -> Result { + let mut guard = self.loaded()?; + let consumption = guard.as_ref().expect("loaded above"); + let now = consumption + .expiry_time_floor + .map_or(now, |floor| now.max(floor)); + let mut updated = consumption.clone(); + let answer = change(&mut updated.volumes, now)?; + if updated.volumes == consumption.volumes { + return Ok(answer); + } + updated.expiry_time_floor = Some(now); + self.write(&mut guard, updated)?; + Ok(answer) + } + + pub(super) fn read_volumes( + &self, + read: impl FnOnce(&VolumeRecords) -> R, + ) -> Result { + let guard = self.loaded()?; + Ok(read(&guard.as_ref().expect("loaded above").volumes)) + } + fn write( &self, guard: &mut std::sync::MutexGuard<'_, Option>, diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs index 59808a5e3c..30a00ccf68 100644 --- a/core/crates/ployzd/src/machine/local_machine/volume_approval.rs +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval.rs @@ -1,14 +1,142 @@ -//! The Volume verbs an enforcing Machine refuses from an Operator caller. A Volume -//! approval a Device signs replaces each refusal; until then a Full caller still runs them. +//! The Volume verbs an enforcing Machine refuses from an Operator caller, and this +//! Machine's own record of the Volume operations a Device approved: which are open, which +//! ended, who may pull what, and which recipients were withdrawn. +//! +//! An operation is admitted once inside its window and then runs on its record, so the +//! work a person started finishes after the window. An ended operation never runs again. +//! Each record is synced before the effect it guards, and none reads a clock again. + +use std::collections::BTreeMap; use ployz_core::{ - SEED_REFUSED_BY_ENFORCEMENT, SEED_VOLUME_CAPABILITY, SET_VOLUME_COPY_CAPABILITY, + CopyTransition, DockerVolumeName, MachineId, OperationId, SEED_REFUSED_BY_ENFORCEMENT, + SEED_VOLUME_CAPABILITY, SET_VOLUME_COPY_CAPABILITY, Sealing, SnapshotName, VOLUME_START_REFUSED_BY_ENFORCEMENT, + authority::{ + AuthorityRefusal, Signed, StepDigest, Timestamp, VolumeApproval, VolumeFacts, VolumeStep, + VolumeWorkflow, volume, + }, }; +use serde::{Deserialize, Serialize}; use super::{Error, LocalMachine}; impl LocalMachine { + /// Admit `step` on `volume` under `approval`. An operation this Machine ended is + /// refused for good. An operation open here passes on its record, at any time, for any + /// step its workflow permits this Machine. Otherwise the approval must verify inside its + /// window, and the operation is recorded open, synced, before this returns. Admitting + /// an operation that owns this Machine's copy ends the one that owned it before. + /// Admitting a replacing Mirror's opening `MirrorToForget` also withdraws this Machine + /// from the Volume in that same write, just before the replacement opens, so the + /// withdrawal ends every other Mirror operation here and none of the replacement's. The + /// caller writes no other withdrawal for that `Forget`. + /// + /// # Errors + /// Returns [`Error::VolumeStepRefused`] naming what the approval lacks, an operation + /// already ended here as [`AuthorityRefusal::Replayed`], and [`Error::ApprovalRecord`] + /// when the record cannot be read or written. + pub fn admit_volume_step( + &self, + approval: &Signed, + volume: &DockerVolumeName, + step: &VolumeStep, + ) -> Result { + let record = self.record(); + let machine = record.id(); + let trusted = record.trusted_keys(); + let cluster = match self.held_sealing(record.phase()) { + Some(Sealing::Held { public_key }) => Some(public_key), + _ => None, + }; + let operation = approval.open().map_err(Error::VolumeStepRefused)?.operation; + let digest = StepDigest::of(&approval.body); + self.approvals + .change_volumes(Timestamp::now(), |records, now| { + records.admit(operation, digest, &machine, volume, step, || { + volume::verify( + approval, + &VolumeFacts { + cluster, + machine: &machine, + trusted: &trusted, + volume, + step, + now, + }, + ) + .map(|approved| approved.body.workflow) + }) + }) + .inspect(|()| tracing::info!(%operation, %volume, ?step, "Volume step admitted"))?; + Ok(operation) + } + + /// End `operation` on this Machine, with its grants. It is refused from then on. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn end_volume_operation(&self, operation: OperationId) -> Result<(), Error> { + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.end(operation); + Ok(()) + }) + } + + /// Let `recipient` pull `scope` of `operation`'s Volume from this Machine while the + /// operation is open. A grant merges into the one the operation already holds there. + /// + /// # Errors + /// Returns [`Error::VolumeStepRefused`] for an operation not open here, a recipient + /// its workflow does not send to, or one withdrawn since this Machine admitted it, and + /// [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn grant_volume( + &self, + operation: OperationId, + recipient: MachineId, + scope: GrantScope, + ) -> Result<(), Error> { + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.grant(operation, recipient, scope) + }) + } + + /// Whether a live grant lets `recipient` pull `snapshot` of `volume` from here. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read. + pub fn volume_grant_live( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + snapshot: &SnapshotName, + ) -> Result { + self.approvals + .read_volumes(|records| records.live(volume, recipient, snapshot)) + } + + /// Withdraw `recipient` from `volume` here, before anything is destroyed: every Mirror + /// operation this Machine admitted for it ends, and so does every grant to it. It needs + /// no approval, because it only narrows what a driver can do. An approval this Machine + /// never admitted is still new work while its window lasts. + /// + /// # Errors + /// Returns [`Error::ApprovalRecord`] when the record cannot be read or written. + pub fn withdraw_volume( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + ) -> Result<(), Error> { + let machine = self.record().id(); + self.approvals + .change_volumes(Timestamp::now(), |records, _| { + records.withdraw(&machine, volume, recipient); + Ok(()) + }) + } + /// Under Device enforcement an Operator's `SetVolumeCopy` that starts the Writer's /// Container is refused before the switch lock. A Create runs a spec the caller /// chose. A Kept start names a stopped Container that nothing on this Machine ties @@ -37,6 +165,269 @@ impl LocalMachine { } } +/// What a grant lets its recipient pull. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum GrantScope { + /// Every snapshot of the Volume: a Mirror's, until it is withdrawn. + Standing, + /// Only these snapshots, which this Machine took under the operation. + Snapshots(Vec), +} + +/// This Machine's Volume operations, grants and withdrawals, kept in the approvals log. +/// No record here is pruned. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub(super) struct VolumeRecords { + /// Raised by each admission and each withdrawal, so a withdrawal outranks exactly the + /// operations admitted before it. It orders events on this Machine and reads no clock. + sequence: u64, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + operations: BTreeMap, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + grants: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + withdrawals: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct OperationRecord { + /// The approval's body as signed, so another body under the same ID is refused. + digest: StepDigest, + volume: DockerVolumeName, + workflow: VolumeWorkflow, + admitted: u64, + state: OperationState, + #[serde(default, skip_serializing_if = "Option::is_none")] + decision: Option, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum OperationState { + Open, + Ended, +} + +/// What the source of a Move did after its cut. Written once. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum Decision { + Resumed, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct Grant { + operation: OperationId, + volume: DockerVolumeName, + recipient: MachineId, + scope: GrantScope, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct Withdrawal { + volume: DockerVolumeName, + recipient: MachineId, + at: u64, +} + +impl VolumeRecords { + pub(super) fn is_empty(&self) -> bool { + *self == Self::default() + } + + fn admit( + &mut self, + operation: OperationId, + digest: StepDigest, + machine: &MachineId, + volume: &DockerVolumeName, + step: &VolumeStep, + verify: impl FnOnce() -> Result, + ) -> Result<(), Error> { + let refused = |field| { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { + field, + })) + }; + if let Some(record) = self.operations.get(&operation) { + if record.state == OperationState::Ended { + return Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { + operation, + })); + } + if record.digest != digest { + return refused("an operation"); + } + if record.volume != *volume { + return refused("a Volume"); + } + if !record.workflow.permits(machine, step) { + return refused("a Volume step"); + } + return Ok(()); + } + let workflow = verify().map_err(Error::VolumeStepRefused)?; + if workflow.owns(machine) { + let owners: Vec = self + .operations + .iter() + .filter(|(_, record)| { + record.state == OperationState::Open + && record.volume == *volume + && record.workflow.owns(machine) + }) + .map(|(operation, _)| *operation) + .collect(); + for owner in owners { + self.end_owner(owner, machine); + } + } + let replaces_here = matches!( + &workflow, + VolumeWorkflow::Mirror { mirror, replace: true, .. } if mirror == machine + ) && *step == VolumeStep::Copy(CopyTransition::MirrorToForget); + if replaces_here { + self.withdraw(machine, volume, machine); + } + self.sequence += 1; + self.operations.insert( + operation, + OperationRecord { + digest, + volume: volume.clone(), + workflow, + admitted: self.sequence, + state: OperationState::Open, + decision: None, + }, + ); + Ok(()) + } + + /// End an earlier owner of this Machine's copy. A Move this Machine is the source of, + /// and that has no decision yet, records `Resumed`, so no handoff can follow. + fn end_owner(&mut self, operation: OperationId, machine: &MachineId) { + if let Some(record) = self.operations.get_mut(&operation) + && record.workflow.moves_from(machine) + && record.decision.is_none() + { + record.decision = Some(Decision::Resumed); + } + self.end(operation); + } + + fn end(&mut self, operation: OperationId) { + if let Some(record) = self.operations.get_mut(&operation) { + record.state = OperationState::Ended; + } + self.grants.retain(|grant| grant.operation != operation); + } + + fn grant( + &mut self, + operation: OperationId, + recipient: MachineId, + scope: GrantScope, + ) -> Result<(), Error> { + let refused = |refusal| Err(Error::VolumeStepRefused(refusal)); + let Some(record) = self.operations.get(&operation) else { + return refused(AuthorityRefusal::Mismatch { + field: "an operation", + }); + }; + if record.state == OperationState::Ended { + return refused(AuthorityRefusal::Replayed { operation }); + } + let target = + matches!(&record.workflow, VolumeWorkflow::Move { to, .. } if *to == recipient); + let sends = record.workflow.mirrors_to(&recipient) + || (target && matches!(scope, GrantScope::Snapshots(_))); + if !sends { + return refused(AuthorityRefusal::Mismatch { + field: "a recipient", + }); + } + let outranked = self.withdrawals.iter().any(|withdrawal| { + withdrawal.volume == record.volume + && withdrawal.recipient == recipient + && withdrawal.at > record.admitted + }); + if outranked { + return refused(AuthorityRefusal::Replayed { operation }); + } + let volume = record.volume.clone(); + let held = self + .grants + .iter_mut() + .find(|grant| grant.operation == operation && grant.recipient == recipient); + match (held, scope) { + (Some(held), GrantScope::Standing) => held.scope = GrantScope::Standing, + (Some(held), GrantScope::Snapshots(added)) => { + if let GrantScope::Snapshots(names) = &mut held.scope { + for name in added { + if !names.contains(&name) { + names.push(name); + } + } + } + } + (None, scope) => self.grants.push(Grant { + operation, + volume, + recipient, + scope, + }), + } + Ok(()) + } + + fn live( + &self, + volume: &DockerVolumeName, + recipient: &MachineId, + snapshot: &SnapshotName, + ) -> bool { + self.grants.iter().any(|grant| { + grant.volume == *volume + && grant.recipient == *recipient + && match &grant.scope { + GrantScope::Standing => true, + GrantScope::Snapshots(names) => names.contains(snapshot), + } + }) + } + + /// End every open operation here that mirrors `volume` to `recipient`: a Mirror's own, + /// on either Machine, and a Move's following on the recipient itself. A Move between + /// two other Machines goes on, but can no longer grant `recipient` anything. + fn withdraw(&mut self, machine: &MachineId, volume: &DockerVolumeName, recipient: &MachineId) { + self.sequence += 1; + self.withdrawals.push(Withdrawal { + volume: volume.clone(), + recipient: *recipient, + at: self.sequence, + }); + let mirrored: Vec = self + .operations + .iter() + .filter(|(_, record)| { + record.state == OperationState::Open + && record.volume == *volume + && record.workflow.mirrors_to(recipient) + && (matches!(record.workflow, VolumeWorkflow::Mirror { .. }) + || machine == recipient) + }) + .map(|(operation, _)| *operation) + .collect(); + for operation in mirrored { + self.end(operation); + } + self.grants + .retain(|grant| !(grant.volume == *volume && grant.recipient == *recipient)); + } +} + /// What a refused Volume verb answers, by the capability refused. pub(super) fn refusal_text(capability: &str) -> &'static str { if capability == SEED_VOLUME_CAPABILITY { @@ -45,3 +436,7 @@ pub(super) fn refusal_text(capability: &str) -> &'static str { VOLUME_START_REFUSED_BY_ENFORCEMENT } } + +#[cfg(test)] +#[path = "volume_approval_tests.rs"] +mod tests; diff --git a/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs new file mode 100644 index 0000000000..6bbfa4bfad --- /dev/null +++ b/core/crates/ployzd/src/machine/local_machine/volume_approval_tests.rs @@ -0,0 +1,850 @@ +use std::path::Path; + +use ployz_core::{ + CopyTransition, DeviceKey, MachineName, Namespace, ServiceName, SnapshotGuid, + authority::{ApprovedService, PayloadDigest, Signer, SignerKey, validity::MAX_LIFETIME_MILLIS}, +}; + +use super::*; +use crate::machine::{ + LocalMachineStore, RecordOwner, local_machine::restart_approval::ConsumptionLog, +}; + +fn id(fill: char) -> MachineId { + MachineId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn data() -> DockerVolumeName { + DockerVolumeName::parse("app_data").unwrap() +} + +fn other_volume() -> DockerVolumeName { + DockerVolumeName::parse("app_other").unwrap() +} + +fn operation(fill: char) -> OperationId { + OperationId::parse(fill.to_string().repeat(32)).unwrap() +} + +fn digest(fill: char) -> StepDigest { + StepDigest::of(&fill.to_string()) +} + +fn at(millis: u64) -> Timestamp { + Timestamp::from_millis(millis).unwrap() +} + +fn mirror(writer: char, mirror: char) -> VolumeWorkflow { + VolumeWorkflow::Mirror { + writer: id(writer), + mirror: id(mirror), + replace: false, + } +} + +fn moving(from: char, to: char, mirrors: &[char]) -> VolumeWorkflow { + moving_from(id(from), to, mirrors) +} + +fn moving_from(from: MachineId, to: char, mirrors: &[char]) -> VolumeWorkflow { + VolumeWorkflow::Move { + from, + to: id(to), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::parse(&"d".repeat(64)).unwrap(), + image: None, + }, + mirrors: mirrors.iter().map(|fill| id(*fill)).collect(), + } +} + +fn restore(machine: char) -> VolumeWorkflow { + VolumeWorkflow::Restore { + machine: id(machine), + at: SnapshotGuid::new(7), + loses_after: at(1), + } +} + +fn snapshot() -> VolumeStep { + VolumeStep::Copy(CopyTransition::WriterToSnapshot) +} + +fn pull() -> VolumeStep { + VolumeStep::Copy(CopyTransition::MirrorToMirror) +} + +fn name(value: &str) -> SnapshotName { + value.parse().unwrap() +} + +/// One Machine's view of the approvals log at `dir`, opened afresh for each call, so every +/// answer comes from what reached the disk. +struct Records<'a> { + dir: &'a Path, + machine: MachineId, +} + +impl Records<'_> { + fn log(&self) -> std::sync::Arc { + ConsumptionLog::new(self.dir) + } + + /// Admit `step` of operation `op` with `body` as its digest. `verified` stands in for the + /// signed approval's check, which only a first step reaches. + fn admit( + &self, + op: char, + body: char, + volume: &DockerVolumeName, + step: &VolumeStep, + verified: Result, + ) -> Result<(), Error> { + self.log().change_volumes(at(5_000), |records, _| { + records.admit( + operation(op), + digest(body), + &self.machine, + volume, + step, + || verified, + ) + }) + } + + fn admit_new(&self, op: char, workflow: VolumeWorkflow, step: &VolumeStep) { + self.admit(op, op, &data(), step, Ok(workflow)).unwrap(); + } + + /// A later step of an open operation, whose window has passed. + fn admit_late(&self, op: char, step: &VolumeStep) -> Result<(), Error> { + self.admit(op, op, &data(), step, Err(AuthorityRefusal::Expired)) + } + + fn grant(&self, op: char, recipient: char, scope: GrantScope) -> Result<(), Error> { + self.log().change_volumes(at(5_000), |records, _| { + records.grant(operation(op), id(recipient), scope) + }) + } + + fn live(&self, volume: &DockerVolumeName, recipient: char, snapshot: &str) -> bool { + self.log() + .read_volumes(|records| records.live(volume, &id(recipient), &name(snapshot))) + .unwrap() + } + + fn withdraw(&self, volume: &DockerVolumeName, recipient: char) { + self.log() + .change_volumes(at(5_000), |records, _| { + records.withdraw(&self.machine, volume, &id(recipient)); + Ok(()) + }) + .unwrap(); + } + + fn withdrawals(&self) -> Vec { + self.log() + .read_volumes(|records| records.withdrawals.clone()) + .unwrap() + } + + fn end(&self, op: char) { + self.log() + .change_volumes(at(5_000), |records, _| { + records.end(operation(op)); + Ok(()) + }) + .unwrap(); + } + + fn decision(&self, op: char) -> Option { + self.log() + .read_volumes(|records| { + records + .operations + .get(&operation(op)) + .and_then(|record| record.decision) + }) + .unwrap() + } +} + +fn replayed(result: Result<(), Error>, op: char) -> bool { + matches!( + result, + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { operation: refused })) + if refused == operation(op) + ) +} + +fn mismatch(result: Result<(), Error>) -> Option<&'static str> { + match result { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { field })) => Some(field), + _ => None, + } +} + +#[test] +fn an_open_operation_finishes_after_its_window_and_an_ended_one_never_runs_again() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + + assert!(matches!( + writer.admit( + '1', + '1', + &data(), + &snapshot(), + Err(AuthorityRefusal::Expired) + ), + Err(Error::VolumeStepRefused(AuthorityRefusal::Expired)) + )); + assert!( + !dir.path().join("restart-approvals.json").exists(), + "a refused first step writes nothing" + ); + + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + writer.admit_late('1', &snapshot()).unwrap(); + writer + .log() + .change_volumes(at(1), |records, _| { + records.admit( + operation('1'), + digest('1'), + &id('a'), + &data(), + &snapshot(), + || unreachable!("an open operation is not checked against the clock"), + ) + }) + .unwrap(); + writer.grant('1', 'b', GrantScope::Standing).unwrap(); + assert!(writer.live(&data(), 'b', "w-1-1")); + + writer.end('1'); + assert!(replayed(writer.admit_late('1', &snapshot()), '1')); + assert!(replayed( + writer.admit('1', '1', &data(), &snapshot(), Ok(mirror('a', 'b'))), + '1' + )); + assert!(!writer.live(&data(), 'b', "w-1-1")); + assert!(replayed(writer.grant('1', 'b', GrantScope::Standing), '1')); +} + +#[test] +fn an_open_operation_still_admits_only_its_own_body_volume_and_graph() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + assert_eq!( + mismatch(writer.admit('1', '2', &data(), &snapshot(), Ok(mirror('a', 'b')))), + Some("an operation") + ); + assert_eq!( + mismatch(writer.admit('1', '1', &other_volume(), &snapshot(), Ok(mirror('a', 'b')))), + Some("a Volume") + ); + assert_eq!( + mismatch(writer.admit_late('1', &VolumeStep::Copy(CopyTransition::WriterToMirror))), + Some("a Volume step") + ); + assert_eq!( + mismatch(writer.grant('1', 'c', GrantScope::Standing)), + Some("a recipient") + ); + assert_eq!( + mismatch(writer.grant('9', 'b', GrantScope::Standing)), + Some("an operation") + ); +} + +#[test] +fn a_bounded_grant_serves_only_its_snapshots_and_ends_with_its_move() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &[]), &snapshot()); + assert_eq!( + mismatch(source.grant('1', 'b', GrantScope::Standing)), + Some("a recipient"), + "a Move's target never holds every snapshot" + ); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("w-1-1")])) + .unwrap(); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + assert!(source.live(&data(), 'b', "w-1-1")); + assert!(source.live(&data(), 'b', "f-1")); + assert!(!source.live(&data(), 'b', "w-2-1")); + assert!(!source.live(&data(), 'c', "f-1")); + assert!(!source.live(&other_volume(), 'b', "f-1")); + + source.end('1'); + assert!(!source.live(&data(), 'b', "f-1")); +} + +#[test] +fn a_withdrawal_ends_only_what_mirrored_that_volume_to_that_machine() { + let dir = tempfile::tempdir().unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + writer.admit_new('2', mirror('a', 'c'), &snapshot()); + writer + .admit('3', '3', &other_volume(), &snapshot(), Ok(mirror('a', 'b'))) + .unwrap(); + for (op, recipient) in [('1', 'b'), ('2', 'c'), ('3', 'b')] { + writer.grant(op, recipient, GrantScope::Standing).unwrap(); + } + + writer.withdraw(&data(), 'b'); + + assert!(replayed(writer.admit_late('1', &snapshot()), '1')); + assert!(replayed( + writer.admit('1', '1', &data(), &snapshot(), Ok(mirror('a', 'b'))), + '1' + )); + assert!(!writer.live(&data(), 'b', "w-1-1")); + writer.admit_late('2', &snapshot()).unwrap(); + assert!(writer.live(&data(), 'c', "w-1-1")); + writer + .admit( + '3', + '3', + &other_volume(), + &snapshot(), + Err(AuthorityRefusal::Expired), + ) + .unwrap(); + assert!(writer.live(&other_volume(), 'b', "w-1-1")); + + writer.admit_new('4', mirror('a', 'b'), &snapshot()); + writer.grant('4', 'b', GrantScope::Standing).unwrap(); + assert!(writer.live(&data(), 'b', "w-1-1")); +} + +#[test] +fn a_mirror_that_withdraws_itself_ends_its_own_pulls_and_a_followed_move() { + let dir = tempfile::tempdir().unwrap(); + let removed = Records { + dir: dir.path(), + machine: id('c'), + }; + removed.admit_new('1', mirror('a', 'c'), &pull()); + removed.withdraw(&data(), 'c'); + assert!(replayed(removed.admit_late('1', &pull()), '1')); + + removed.admit_new('2', moving('a', 'b', &['c']), &pull()); + removed.withdraw(&data(), 'c'); + assert!(replayed(removed.admit_late('2', &pull()), '2')); +} + +#[test] +fn a_withdrawal_outranks_a_grant_under_a_move_admitted_before_it() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &['c']), &snapshot()); + source.withdraw(&data(), 'c'); + + source.admit_late('1', &snapshot()).unwrap(); + assert!(replayed(source.grant('1', 'c', GrantScope::Standing), '1')); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + + source.admit_new('2', moving('a', 'b', &['c']), &snapshot()); + source.grant('2', 'c', GrantScope::Standing).unwrap(); +} + +#[test] +fn a_new_owner_of_the_copy_ends_the_old_one_and_a_source_without_a_decision_resumes() { + let dir = tempfile::tempdir().unwrap(); + let source = Records { + dir: dir.path(), + machine: id('a'), + }; + source.admit_new('1', moving('a', 'b', &[]), &snapshot()); + source + .grant('1', 'b', GrantScope::Snapshots(vec![name("f-1")])) + .unwrap(); + source.admit_new('2', mirror('e', 'a'), &pull()); + source.admit_new( + '3', + VolumeWorkflow::Resize { + machines: vec![id('a')], + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + shrink: false, + }, + &VolumeStep::Resize { + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + allow_shrink: false, + }, + ); + assert!(replayed(source.admit_late('1', &snapshot()), '1')); + assert_eq!(source.decision('1'), Some(Decision::Resumed)); + assert!(!source.live(&data(), 'b', "f-1")); + + source.admit_new( + '4', + restore('a'), + &VolumeStep::Copy(CopyTransition::MirrorToWriter), + ); + assert!(replayed(source.admit_late('2', &pull()), '2')); + assert_eq!(source.decision('2'), None); + source + .admit_late( + '3', + &VolumeStep::Resize { + maximum_bytes: ployz_core::ProvisionedVolumeMaximumBytes::new( + std::num::NonZeroU64::new(1_000_000_000).unwrap(), + ), + allow_shrink: false, + }, + ) + .unwrap(); + + let target = Records { + dir: dir.path(), + machine: id('b'), + }; + target + .admit( + '5', + '5', + &other_volume(), + &pull(), + Ok(moving('a', 'b', &[])), + ) + .unwrap(); + target + .admit( + '6', + '6', + &other_volume(), + &VolumeStep::Copy(CopyTransition::MirrorToWriter), + Ok(restore('b')), + ) + .unwrap(); + assert!(replayed( + target.admit( + '5', + '5', + &other_volume(), + &pull(), + Err(AuthorityRefusal::Expired) + ), + '5' + )); + assert_eq!(target.decision('5'), None, "b is not the Move's source"); +} + +#[test] +fn records_written_before_this_format_existed_read_back_with_no_volume_records() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("restart-approvals.json"), + br#"{"expiry_time_floor":1000}"#, + ) + .unwrap(); + let writer = Records { + dir: dir.path(), + machine: id('a'), + }; + assert!(!writer.live(&data(), 'b', "w-1-1")); + writer.admit_new('1', mirror('a', 'b'), &snapshot()); + let stored = std::fs::read_to_string(dir.path().join("restart-approvals.json")).unwrap(); + assert!( + stored.contains(r#""volumes":{"sequence":1,"operations""#), + "{stored}" + ); +} + +struct Enforcing { + machine: LocalMachine, + signer: Signer, + cluster: ployz_core::SealingPublicKey, + data_dir: crate::test_dir::TestDir, +} + +async fn enforcing() -> Enforcing { + let sealing = crate::sealed_env::test_support::Cluster::new(); + let data_dir = crate::test_dir::TestDir::new("volume-approval"); + let mut store = LocalMachineStore::open(&data_dir.0).unwrap(); + store + .enroll(ployz_core::EnrollRequest { + role: ployz_core::EnrollAs::Founder { + initial_policy: Default::default(), + name: MachineName::parse("local").unwrap(), + cluster_network: "10.210.0.0/16".parse().unwrap(), + public_ip: None, + advertised_endpoints: vec![ployz_core::AdvertisedEndpoint( + "192.0.2.1:51820".parse().unwrap(), + )], + grant: None, + }, + wireguard_mtu: None, + pinned: Default::default(), + }) + .unwrap(); + std::fs::copy( + sealing.dir.0.join("sealing.key"), + data_dir.0.join("sealing.key"), + ) + .unwrap(); + let owner = RecordOwner::spawn(store).unwrap(); + let signer = Signer::from_seed(&[7; 32]); + let device = DeviceKey::from(signer.public()); + owner + .mutate(move |store| { + store.approve_device(device).unwrap(); + store.set_device_enforcement(true, Vec::new()).unwrap(); + }) + .await + .unwrap(); + let cluster = crate::sealing_key::SealingKey::load(&data_dir.0) + .unwrap() + .unwrap() + .public_key(); + Enforcing { + machine: LocalMachine::new(owner), + signer, + cluster, + data_dir, + } +} + +impl Enforcing { + /// A Move of `app_data` from this Machine to another, signed now by `signer`, open + /// from `issued` milliseconds ago for `lifetime` milliseconds. + fn approval(&self, signer: &Signer, issued: u64, lifetime: u64) -> Signed { + let issued_at = at(Timestamp::now().as_millis() - issued); + let body = VolumeApproval { + cluster: self.cluster, + operation: operation('1'), + namespace: Namespace::parse("app").unwrap(), + volume: data(), + workflow: moving_from(self.machine.record().id(), 'b', &[]), + issued_at, + expires_at: issued_at.plus(lifetime), + signer: SignerKey::Device(DeviceKey::from(signer.public())), + }; + Signed::sign(&body, signer).unwrap() + } + + /// `workflow` on `volume`, signed now by this Machine's Device. + fn signed(&self, volume: DockerVolumeName, workflow: VolumeWorkflow) -> Signed { + let issued_at = Timestamp::now(); + let body = VolumeApproval { + cluster: self.cluster, + operation: operation('1'), + namespace: Namespace::parse("app").unwrap(), + volume, + workflow, + issued_at, + expires_at: issued_at.plus(MAX_LIFETIME_MILLIS), + signer: SignerKey::Device(DeviceKey::from(self.signer.public())), + }; + Signed::sign(&body, &self.signer).unwrap() + } + + fn records_empty(&self) -> bool { + ConsumptionLog::new(&self.data_dir.0) + .read_volumes(VolumeRecords::is_empty) + .unwrap() + } +} + +#[tokio::test] +async fn only_a_replacing_mirror_of_this_volume_and_machine_withdraws_with_its_forget() { + let fixture = enforcing().await; + let here = fixture.machine.record().id(); + let mirror_here = |replace| VolumeWorkflow::Mirror { + writer: id('a'), + mirror: here, + replace, + }; + let refused = |volume: DockerVolumeName, workflow: VolumeWorkflow| match fixture + .machine + .admit_volume_step(&fixture.signed(volume, workflow), &data(), &forget()) + { + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { field })) => Some(field), + _ => None, + }; + + assert_eq!(refused(data(), mirror_here(false)), Some("a Volume step")); + assert_eq!( + refused( + data(), + VolumeWorkflow::Move { + from: id('a'), + to: id('b'), + service: ApprovedService { + name: ServiceName::parse("db").unwrap(), + digest: PayloadDigest::parse(&"d".repeat(64)).unwrap(), + image: None, + }, + mirrors: vec![here], + } + ), + Some("a Volume step") + ); + assert_eq!(refused(data(), replacing('a', 'c')), Some("this Server")); + assert_eq!(refused(other_volume(), mirror_here(true)), Some("a Volume")); + assert!(fixture.records_empty(), "a refused Forget writes nothing"); + + let replacement = fixture.signed(data(), mirror_here(true)); + fixture + .machine + .admit_volume_step(&replacement, &data(), &forget()) + .unwrap(); + fixture + .machine + .admit_volume_step(&replacement, &data(), &forget()) + .unwrap(); + let withdrawals = ConsumptionLog::new(&fixture.data_dir.0) + .read_volumes(|records| records.withdrawals.clone()) + .unwrap(); + assert_eq!( + withdrawals, + vec![Withdrawal { + volume: data(), + recipient: here, + at: 1, + }] + ); +} + +#[tokio::test] +async fn a_device_approval_opens_the_operation_on_disk_and_a_late_or_foreign_one_is_refused() { + let fixture = enforcing().await; + let late = fixture.approval( + &fixture.signer, + MAX_LIFETIME_MILLIS + 600_000, + MAX_LIFETIME_MILLIS, + ); + assert!(matches!( + fixture + .machine + .admit_volume_step(&late, &data(), &snapshot()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Expired)) + )); + let stranger = Signer::from_seed(&[3; 32]); + assert!(matches!( + fixture.machine.admit_volume_step( + &fixture.approval(&stranger, 0, MAX_LIFETIME_MILLIS), + &data(), + &snapshot() + ), + Err(Error::VolumeStepRefused( + AuthorityRefusal::UnknownKey { .. } + )) + )); + + let approval = fixture.approval(&fixture.signer, 0, MAX_LIFETIME_MILLIS); + assert_eq!( + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()) + .unwrap(), + operation('1') + ); + assert!(matches!( + fixture + .machine + .admit_volume_step(&approval, &data(), &pull()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Mismatch { + field: "a Volume step" + })) + )); + fixture + .machine + .grant_volume( + operation('1'), + id('b'), + GrantScope::Snapshots(vec![name("f-1")]), + ) + .unwrap(); + assert!( + fixture + .machine + .volume_grant_live(&data(), &id('b'), &name("f-1")) + .unwrap() + ); + + fixture + .machine + .end_volume_operation(operation('1')) + .unwrap(); + let reopened = ConsumptionLog::new(&fixture.data_dir.0); + assert!( + !reopened + .read_volumes(|records| records.live(&data(), &id('b'), &name("f-1"))) + .unwrap() + ); + assert!(matches!( + reopened.change_volumes(Timestamp::now(), |records, _| { + records.admit( + operation('1'), + StepDigest::of(&approval.body), + &fixture.machine.record().id(), + &data(), + &snapshot(), + || unreachable!("an ended operation is refused before any check"), + ) + }), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); + assert!(matches!( + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); +} + +#[tokio::test] +async fn a_withdrawal_by_this_machine_survives_a_reload() { + let fixture = enforcing().await; + let approval = fixture.approval(&fixture.signer, 0, MAX_LIFETIME_MILLIS); + fixture + .machine + .admit_volume_step(&approval, &data(), &snapshot()) + .unwrap(); + fixture.machine.withdraw_volume(&data(), &id('b')).unwrap(); + assert!(matches!( + fixture.machine.grant_volume( + operation('1'), + id('b'), + GrantScope::Snapshots(vec![name("f-1")]) + ), + Err(Error::VolumeStepRefused(AuthorityRefusal::Replayed { .. })) + )); + let reopened = ConsumptionLog::new(&fixture.data_dir.0); + let withdrawn = reopened + .read_volumes(|records| records.withdrawals.clone()) + .unwrap(); + assert_eq!( + withdrawn, + vec![Withdrawal { + volume: data(), + recipient: id('b'), + at: 2, + }] + ); +} + +#[test] +fn a_later_owner_of_a_following_mirrors_copy_ends_the_move_there() { + let dir = tempfile::tempdir().unwrap(); + let follower = Records { + dir: dir.path(), + machine: id('c'), + }; + follower.admit_new('1', moving('a', 'b', &['c']), &pull()); + follower.admit_new('2', mirror('b', 'c'), &pull()); + + assert!(replayed(follower.admit_late('1', &pull()), '1')); + follower.admit_late('2', &pull()).unwrap(); +} + +fn replacing(writer: char, mirror: char) -> VolumeWorkflow { + VolumeWorkflow::Mirror { + writer: id(writer), + mirror: id(mirror), + replace: true, + } +} + +fn forget() -> VolumeStep { + VolumeStep::Copy(CopyTransition::MirrorToForget) +} + +#[test] +fn a_replacing_mirrors_opening_forget_withdraws_every_other_mirror_here_and_keeps_itself() { + let dir = tempfile::tempdir().unwrap(); + let replaced = Records { + dir: dir.path(), + machine: id('b'), + }; + replaced.admit_new('1', mirror('a', 'b'), &pull()); + replaced + .admit('3', '3', &other_volume(), &pull(), Ok(mirror('a', 'b'))) + .unwrap(); + + replaced.admit_new('2', replacing('a', 'b'), &forget()); + + assert_eq!( + replaced.withdrawals(), + vec![Withdrawal { + volume: data(), + recipient: id('b'), + at: 3, + }] + ); + assert!(replayed(replaced.admit_late('1', &pull()), '1')); + replaced + .admit_late('2', &VolumeStep::Copy(CopyTransition::NoneToMirror)) + .unwrap(); + replaced + .admit( + '3', + '3', + &other_volume(), + &pull(), + Err(AuthorityRefusal::Expired), + ) + .unwrap(); +} + +#[test] +fn a_replacement_withdraws_once_and_a_later_withdrawal_ends_it() { + let dir = tempfile::tempdir().unwrap(); + let replaced = Records { + dir: dir.path(), + machine: id('b'), + }; + replaced.admit_new('2', replacing('a', 'b'), &forget()); + replaced.admit_late('2', &forget()).unwrap(); + assert_eq!( + replaced.withdrawals().len(), + 1, + "a resent Forget withdraws nothing" + ); + replaced.admit_late('2', &pull()).unwrap(); + + replaced.withdraw(&data(), 'b'); + assert!(replayed(replaced.admit_late('2', &pull()), '2')); +} + +#[test] +fn a_replacement_opened_by_a_pull_withdraws_nothing() { + let dir = tempfile::tempdir().unwrap(); + let mirror_machine = Records { + dir: dir.path(), + machine: id('b'), + }; + mirror_machine.admit_new('2', replacing('a', 'b'), &pull()); + assert!(mirror_machine.withdrawals().is_empty()); + mirror_machine.admit_late('2', &pull()).unwrap(); +} diff --git a/core/crates/ployzd/src/machine_api/local.rs b/core/crates/ployzd/src/machine_api/local.rs index 2a8005d3c5..ecf88538e4 100644 --- a/core/crates/ployzd/src/machine_api/local.rs +++ b/core/crates/ployzd/src/machine_api/local.rs @@ -1288,6 +1288,13 @@ fn local_failure(error: LocalMachineError) -> Result { "next": "ployz device enforce --off from an approved device", }), )), + LocalMachineError::VolumeStepRefused(_) => Err(ployz_core::rpc::hinted_status( + tonic::Code::PermissionDenied, + &error, + serde_json::json!({ + "next": "approve the Volume change again from an approved device", + }), + )), LocalMachineError::RuleGrantRefused(_) | LocalMachineError::RuleProofRefused(_) | LocalMachineError::RuleHeadUnread(_) => Err(build_refused( diff --git a/dashboard/src/modules/approval-session/volume-approval.test.ts b/dashboard/src/modules/approval-session/volume-approval.test.ts new file mode 100644 index 0000000000..c3eeeab997 --- /dev/null +++ b/dashboard/src/modules/approval-session/volume-approval.test.ts @@ -0,0 +1,49 @@ +import type { MachineId, OperationId } from "@ployz/sdk"; +import type { VolumeApproval, VolumeWorkflow } from "@ployz/sdk/authority"; +import { describe, expect, it } from "vitest"; +import { canonicalJson } from "./approval-session"; + +const a = "a".repeat(32) as MachineId; +const b = "b".repeat(32) as MachineId; +const c = "c".repeat(32) as MachineId; +const d = "d".repeat(64); + +const head = `{"cluster":"CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=","expires_at":1000000,"issued_at":100000,"namespace":"app","operation":"0f8fad5bd9cb469fa16570867728950e","signer":"ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c","volume":"app_data","workflow":`; + +// The bodies `ployz_core::authority::volume`'s golden test pins, one per workflow. +const workflows: [VolumeWorkflow, string][] = [ + [ + { kind: "move", from: a, to: b, service: { name: "db", digest: d }, mirrors: [c] }, + `{"from":"${a}","kind":"move","mirrors":["${c}"],"service":{"digest":"${d}","name":"db"},"to":"${b}"}`, + ], + [{ kind: "mirror", writer: a, mirror: b, replace: false }, `{"kind":"mirror","mirror":"${b}","replace":false,"writer":"${a}"}`], + [ + { kind: "mirror_removal", mirror: b, writer: a, role: "mirror", last_copy: true }, + `{"kind":"mirror_removal","last_copy":true,"mirror":"${b}","role":"mirror","writer":"${a}"}`, + ], + [ + { kind: "restore", machine: b, at: "18446744073709551615", loses_after: 500000 }, + `{"at":"18446744073709551615","kind":"restore","loses_after":500000,"machine":"${b}"}`, + ], + [ + { kind: "resize", machines: [a, b], maximum_bytes: 20_000_000_000, shrink: false }, + `{"kind":"resize","machines":["${a}","${b}"],"maximum_bytes":20000000000,"shrink":false}`, + ], + [{ kind: "seed", machine: a, source: "app_old", at: "42" }, `{"at":"42","kind":"seed","machine":"${a}","source":"app_old"}`], +]; + +describe("a Volume approval's canonical body", () => { + it.each(workflows)("matches ployz-core's bytes for %j", (workflow, golden) => { + const approval = { + cluster: "CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=", + operation: "0f8fad5bd9cb469fa16570867728950e" as OperationId, + namespace: "app", + volume: "app_data", + workflow, + issued_at: 100000, + expires_at: 1000000, + signer: "ea4a6c63e29c520abef5507b132ec5f9954776aebebe7b92421eea691446d22c", + } satisfies VolumeApproval; + expect(canonicalJson(approval)).toBe(`${head}${golden}}`); + }); +});