aven_core/sync/client/
enrollment.rs

1//! Repeatable device enrollment and published snapshot retrieval.
2//! The public mailbox never exposes bootstrap chunks, images or credentials.
3mod management;
4use anyhow::{Context as _, Result, ensure};
5use aven_protocol::refusal::Enrollment as Refusal;
6
7use super::errors::is_stale;
8use super::exchange::{self, Link};
9use super::keys::ProtectedLocalKeyStore;
10use super::keys::peer::{ActiveInputs, OpenInvitation};
11use crate::db::Database;
12use crate::sync::seed_claim::{
13    Secret,
14    membership::{self, Evidence, Invitation, Mailbox},
15    peer,
16};
17pub use management::RemovalStatus;
18
19pub struct CreatedInvitation {
20    pub invitation: Invitation,
21    pub state: OpenInvitation,
22    pub vault: [u8; 32],
23    pub resumed: bool,
24}
25
26pub use aven_protocol::wire::enrollment::Context;
27pub use aven_protocol::wire::enrollment::{CONTROL_LIMIT, PATH, PUBLISHED_RESPONSE_LIMIT};
28
29impl From<&ActiveInputs> for Context {
30    fn from(inputs: &ActiveInputs) -> Self {
31        Self {
32            vault: inputs.membership.genesis().context().vault_id,
33            genesis: inputs.membership.genesis().commitment(),
34            device: inputs.device(),
35            head: inputs.membership.head(),
36        }
37    }
38}
39
40pub use aven_protocol::wire::enrollment::Operation;
41pub type Reply =
42    aven_protocol::wire::enrollment::Reply<Evidence, Mailbox, membership::ManagementPreparation>;
43/// One bounded exchange at a time; callers explicitly retry the same protected intent.
44pub struct Client {
45    link: Link,
46    endpoint: url::Url,
47    locator: String,
48}
49impl Client {
50    pub fn new(origin: &str, link: Link) -> Result<Self> {
51        ensure!(
52            origin.len() <= super::MAX_SERVER_BYTES,
53            "error enrollment-locator-limit"
54        );
55        Ok(Self {
56            link,
57            endpoint: super::origin::endpoint(origin, PATH)?,
58            locator: origin.into(),
59        })
60    }
61    pub(crate) fn link(&self) -> &Link {
62        &self.link
63    }
64    pub async fn exchange(&self, op: Operation, secret: Option<&Secret>) -> Result<Reply> {
65        let response_limit = if matches!(&op, Operation::Published { .. }) {
66            PUBLISHED_RESPONSE_LIMIT
67        } else if matches!(&op, Operation::PrepareManagement { .. }) {
68            membership::MAX_EVIDENCE_JSON_BYTES + 128
69        } else if matches!(&op, Operation::Membership { .. }) {
70            membership::MAX_EVIDENCE_JSON_BYTES
71        } else {
72            CONTROL_LIMIT
73        };
74        let bytes =
75            serde_json::to_vec(&op).map_err(|_| anyhow::anyhow!("error enrollment-http"))?;
76        ensure!(bytes.len() <= CONTROL_LIMIT, "error enrollment-limit");
77        let bytes = exchange::post_json(&self.link, &self.endpoint, secret, bytes, response_limit)
78            .await
79            .map_err(|failure| match failure {
80                exchange::Failure::Hosting(hosting) => super::errors::hosting_error(hosting),
81                exchange::Failure::Network => {
82                    anyhow::anyhow!("error enrollment-network outcome-unknown")
83                }
84                exchange::Failure::SecureTransport => {
85                    anyhow::anyhow!("error enrollment-tls outcome-unknown")
86                }
87                exchange::Failure::Malformed => {
88                    anyhow::anyhow!("error enrollment-server outcome-unknown")
89                }
90                exchange::Failure::TooLarge => anyhow::anyhow!("error enrollment-limit"),
91                exchange::Failure::RequestBodyLimit => {
92                    anyhow::anyhow!("error sync-request-body-limit")
93                }
94                // Only an authentication refusal says anything about this
95                // device's access; timeouts and server failures stay
96                // ordinary errors.
97                exchange::Failure::Refused { status, code } => {
98                    match Refusal::classify(status, code.as_deref()) {
99                        Refusal::Stale => membership::StaleContext.into(),
100                        Refusal::Busy => anyhow::anyhow!("error enrollment-busy"),
101                        Refusal::Unauthorized => membership::Unauthorized.into(),
102                        Refusal::Timeout => {
103                            anyhow::anyhow!("error enrollment-timeout outcome-unknown")
104                        }
105                        Refusal::Refused => {
106                            anyhow::anyhow!("error enrollment-refused outcome-unknown")
107                        }
108                        Refusal::Server => {
109                            anyhow::anyhow!("error enrollment-server outcome-unknown")
110                        }
111                    }
112                }
113            })?;
114        serde_json::from_slice(&bytes).map_err(|_| anyhow::anyhow!("error enrollment-http"))
115    }
116    /// Installs only a protected, independently enrolled peer into a fresh target.
117    /// A completed retry is local and never rewinds later changes.
118    pub async fn install(
119        &self,
120        store: &ProtectedLocalKeyStore,
121        db: &Database,
122    ) -> Result<crate::sync::SharedStateInstallReport> {
123        self.install_reporting(store, db, &|_, _| {}).await
124    }
125
126    /// [`Self::install`], reporting downloaded snapshot bytes. The total is
127    /// `None` until the catalogs that size the rest have arrived.
128    pub async fn install_reporting(
129        &self,
130        store: &ProtectedLocalKeyStore,
131        db: &Database,
132        downloaded: &(dyn Fn(u64, Option<u64>) + Sync),
133    ) -> Result<crate::sync::SharedStateInstallReport> {
134        store
135            .install_peer_snapshot(db, self, &self.locator, downloaded)
136            .await
137    }
138
139    #[allow(clippy::too_many_arguments)]
140    pub(crate) async fn download(
141        &self,
142        store: &ProtectedLocalKeyStore,
143        db: &Database,
144        identity: [u8; 32],
145        peer: &membership::Joiner,
146        verified: &membership::VerifiedEnrollment,
147        descriptor: &[u8],
148        downloaded: &(dyn Fn(u64, Option<u64>) + Sync),
149    ) -> Result<crate::sync::bootstrap_format::download::Metadata> {
150        use crate::sync::{
151            bootstrap_format::{self, download::Metadata},
152            bootstrap_staging::{Component, MAX_CHUNKS, MAX_STORAGE_BYTES},
153        };
154        let b = verified.publication().binding();
155        let mut floor = verified.membership().clone();
156        let initial_context = Context {
157            vault: peer.vault(),
158            genesis: verified.genesis().commitment(),
159            device: peer.device(),
160            head: floor.head(),
161        };
162        let (evidence, current) = self.membership(&initial_context, peer.bearer()).await?;
163        floor = store
164            .adopt_download_refresh(db, identity, peer, verified, &evidence, current)
165            .await?;
166        let mut retried = false;
167        let mut read = async |component, index| -> Result<Vec<u8>> {
168            let mut context = Context {
169                vault: peer.vault(),
170                genesis: verified.genesis().commitment(),
171                device: peer.device(),
172                head: floor.head(),
173            };
174            loop {
175                match self
176                    .exchange(
177                        Operation::Published {
178                            context: context.clone(),
179                            descriptor: b.descriptor_commitment,
180                            component,
181                            index,
182                        },
183                        Some(peer.bearer()),
184                    )
185                    .await
186                {
187                    Ok(Reply::Published(bytes)) => return Ok(bytes),
188                    Err(error) if is_stale(&error) && !retried => {
189                        retried = true;
190                        let (evidence, current) = self.membership(&context, peer.bearer()).await?;
191                        floor = store
192                            .adopt_download_refresh(
193                                db, identity, peer, verified, &evidence, current,
194                            )
195                            .await?;
196                        context.head = floor.head();
197                    }
198                    Err(error) => return Err(error),
199                    _ => anyhow::bail!("error snapshot-response"),
200                }
201            }
202        };
203        ensure!(
204            read(None, 0).await? == descriptor,
205            "error snapshot-descriptor-substitution"
206        );
207        let mut package = Metadata {
208            descriptor: descriptor.to_vec(),
209            catalogs: Default::default(),
210            state: vec![],
211            manifest: vec![],
212        };
213        let mut total = 0_u64;
214        let mut chunks = 0_u64;
215        downloaded(0, None);
216        for (i, recipe) in bootstrap_format::download::catalogs(descriptor)?
217            .into_iter()
218            .enumerate()
219        {
220            for (index, length) in recipe.lengths.into_iter().enumerate() {
221                let bytes = read(Some(recipe.component), u64::try_from(index)?).await?;
222                ensure!(
223                    u64::try_from(bytes.len())? == length,
224                    "error snapshot-length"
225                );
226                total = total
227                    .checked_add(length)
228                    .ok_or_else(|| anyhow::anyhow!("error snapshot-limit"))?;
229                chunks += 1;
230                ensure!(
231                    total <= MAX_STORAGE_BYTES && chunks <= MAX_CHUNKS,
232                    "error snapshot-limit"
233                );
234                package.catalogs[i].extend(bytes);
235                downloaded(total, None);
236                crate::sync::crash::Crash::Snapshot.at("download");
237            }
238        }
239        let artifacts: Vec<_> =
240            bootstrap_format::download::artifacts(descriptor, &package.catalogs)?
241                .into_iter()
242                .filter(|recipe| !matches!(recipe.component, Component::Image(_)))
243                .collect();
244        let expected = artifacts
245            .iter()
246            .flat_map(|recipe| &recipe.lengths)
247            .fold(total, |sum, length| sum.saturating_add(*length));
248        downloaded(total, Some(expected));
249        for recipe in artifacts {
250            let mut records = Vec::new();
251            for (index, length) in recipe.lengths.into_iter().enumerate() {
252                let bytes = read(Some(recipe.component), u64::try_from(index)?).await?;
253                ensure!(
254                    u64::try_from(bytes.len())? == length,
255                    "error snapshot-length"
256                );
257                total = total
258                    .checked_add(length)
259                    .ok_or_else(|| anyhow::anyhow!("error snapshot-limit"))?;
260                chunks += 1;
261                ensure!(
262                    total <= MAX_STORAGE_BYTES && chunks <= MAX_CHUNKS,
263                    "error snapshot-limit"
264                );
265                downloaded(total, Some(expected));
266                records.push(bytes);
267            }
268            match recipe.component {
269                Component::Manifest => package.manifest = records,
270                Component::State => package.state = records,
271                _ => anyhow::bail!("error snapshot-component"),
272            }
273        }
274        Ok(package)
275    }
276    async fn membership(
277        &self,
278        context: &Context,
279        bearer: &Secret,
280    ) -> Result<(Evidence, membership::Membership)> {
281        let evidence = self.unverified_membership(context, bearer).await?;
282        let membership = evidence.verify()?;
283        Ok((evidence, membership))
284    }
285    /// The caller must verify the returned chain before use.
286    async fn unverified_membership(&self, context: &Context, bearer: &Secret) -> Result<Evidence> {
287        let Reply::Membership(evidence) = self
288            .exchange(
289                Operation::Membership {
290                    context: context.clone(),
291                },
292                Some(bearer),
293            )
294            .await?
295        else {
296            anyhow::bail!("error membership-response");
297        };
298        Ok(evidence)
299    }
300    pub async fn refresh_inputs(
301        &self,
302        store: &ProtectedLocalKeyStore,
303        db: &Database,
304        inputs: &mut ActiveInputs,
305    ) -> Result<()> {
306        let (evidence, membership) = self
307            .membership(&Context::active(inputs), inputs.bearer())
308            .await?;
309        store
310            .adopt_verified_refresh(db, inputs, evidence, membership)
311            .await
312    }
313    pub async fn refresh(&self, store: &ProtectedLocalKeyStore, db: &Database) -> Result<()> {
314        let mut inputs = store.active_inputs(db, &self.locator).await?;
315        self.refresh_inputs(store, db, &mut inputs).await
316    }
317    #[cfg(any(test, feature = "test-support"))]
318    pub async fn invite(
319        &self,
320        store: &ProtectedLocalKeyStore,
321        db: &Database,
322        expires: u64,
323    ) -> Result<Invitation> {
324        Ok(self
325            .invite_with_status(store, db, expires)
326            .await?
327            .invitation)
328    }
329
330    pub async fn invite_with_status(
331        &self,
332        store: &ProtectedLocalKeyStore,
333        db: &Database,
334        expires: u64,
335    ) -> Result<CreatedInvitation> {
336        let mut inputs = store.active_inputs(db, &self.locator).await?;
337        self.refresh_inputs(store, db, &mut inputs).await?;
338        let previous = store.open_invitation(db, &inputs).await?;
339        let journal = store
340            .prepare_invitation(db, &inputs, Some(expires), None)
341            .await?;
342        let reply = retry_stale!(
343            self.exchange(
344                Operation::Register {
345                    context: Context::active(&inputs),
346                    declaration: journal.declaration.clone(),
347                },
348                Some(inputs.bearer()),
349            )
350            .await,
351            self.refresh_inputs(store, db, &mut inputs).await,
352        )?;
353        let Reply::Registered(peer::RegistrationStatus::Open) = reply else {
354            anyhow::bail!("error enrollment-invitation-unavailable");
355        };
356        let invitation = store.registered_invitation(db, &inputs, &journal).await?;
357        let state = store
358            .open_invitation(db, &inputs)
359            .await?
360            .context("error enrollment-invitation-missing")?;
361        Ok(CreatedInvitation {
362            invitation,
363            state,
364            vault: inputs.membership.genesis().context().vault_id,
365            resumed: previous.is_some_and(|old| old.handle == journal.handle),
366        })
367    }
368    #[cfg(any(test, feature = "test-support"))]
369    pub async fn request(
370        &self,
371        store: &ProtectedLocalKeyStore,
372        db: &Database,
373        invitation: Option<Invitation>,
374    ) -> Result<()> {
375        let peer = self.prepare(store, db, invitation, false).await?;
376        self.post(&peer).await
377    }
378    /// Requests admission with a replacement invitation from the same inviter
379    /// while joining is unfinished. Earlier attempts stay retained and can
380    /// still complete the join.
381    #[cfg(any(test, feature = "test-support"))]
382    pub async fn replace(
383        &self,
384        store: &ProtectedLocalKeyStore,
385        db: &Database,
386        invitation: Invitation,
387    ) -> Result<()> {
388        let peer = self.prepare(store, db, Some(invitation), true).await?;
389        self.post(&peer).await
390    }
391    /// Selects or creates the attempt to post from protected local state.
392    /// Its failures are integrity or eligibility refusals, never outcomes of
393    /// an exchange.
394    pub async fn prepare(
395        &self,
396        store: &ProtectedLocalKeyStore,
397        db: &Database,
398        invitation: Option<Invitation>,
399        replace: bool,
400    ) -> Result<membership::Joiner> {
401        match invitation {
402            Some(invitation) if replace => store.replace_peer(db, &self.locator, invitation).await,
403            invitation => store.prepare_peer(db, &self.locator, invitation).await,
404        }
405    }
406    /// Whether attempts other than the one `prepare` returned could still
407    /// complete this join.
408    pub async fn has_other_attempts(
409        &self,
410        store: &ProtectedLocalKeyStore,
411        db: &Database,
412    ) -> Result<bool> {
413        Ok(store.peer_attempts(db, &self.locator).await?.len() > 1)
414    }
415    pub async fn post(&self, peer: &membership::Joiner) -> Result<()> {
416        ensure!(
417            matches!(
418                self.exchange(
419                    Operation::Post {
420                        vault: peer.vault(),
421                        handle: peer.handle(),
422                        request: peer.request().to_vec()
423                    },
424                    None
425                )
426                .await?,
427                Reply::Done
428            ),
429            "error enrollment-response"
430        );
431        Ok(())
432    }
433    #[cfg(any(test, feature = "test-support"))]
434    pub async fn admit(&self, store: &ProtectedLocalKeyStore, db: &Database) -> Result<bool> {
435        self.admit_handle(store, db, None).await
436    }
437    /// Whether a device has asked to join with this invitation. Only the
438    /// server is consulted, so frequent checks stay cheap.
439    pub async fn join_requested(&self, vault: [u8; 32], handle: [u8; 32]) -> Result<bool> {
440        match self
441            .exchange(Operation::Mailbox { vault, handle }, None)
442            .await?
443        {
444            Reply::Mailbox(mail) => Ok(mail.request.is_some()),
445            _ => anyhow::bail!("error enrollment-response"),
446        }
447    }
448    /// Exact historical outcomes remain addressable after subsequent invitations.
449    pub async fn admit_handle(
450        &self,
451        store: &ProtectedLocalKeyStore,
452        db: &Database,
453        handle: Option<[u8; 32]>,
454    ) -> Result<bool> {
455        let mut inputs = store.active_inputs(db, &self.locator).await?;
456        self.refresh_inputs(store, db, &mut inputs).await?;
457        let journal = store.prepare_invitation(db, &inputs, None, handle).await?;
458        let Reply::Mailbox(mail) = self
459            .exchange(
460                Operation::Mailbox {
461                    vault: inputs.membership.genesis().context().vault_id,
462                    handle: journal.handle,
463                },
464                None,
465            )
466            .await?
467        else {
468            anyhow::bail!("error enrollment-response");
469        };
470        let Some(request) = mail.request else {
471            return Ok(false);
472        };
473        let (record, reply) = retry_stale!(
474            async {
475                let record = store
476                    .prepare_admission(db, &inputs, &journal, &request)
477                    .await?;
478                let reply = self
479                    .exchange(
480                        Operation::Admit {
481                            context: Context::active(&inputs),
482                            handle: journal.handle,
483                            record: record.clone(),
484                        },
485                        Some(inputs.bearer()),
486                    )
487                    .await?;
488                anyhow::Ok((record, reply))
489            }
490            .await,
491            self.refresh_inputs(store, db, &mut inputs).await,
492        )?;
493        let Reply::Admitted(accepted) = reply else {
494            anyhow::bail!("error enrollment-response");
495        };
496        ensure!(accepted == record, "error enrollment-outcome-mismatch");
497        self.refresh_inputs(store, db, &mut inputs).await?;
498        store.finish_inviter(db, &inputs, &journal, &record).await?;
499        Ok(true)
500    }
501    /// Checks every retained attempt for an admission and completes the one
502    /// whose grant opens for its exact request; a grant that does not open is
503    /// an integrity failure. A refusal or unreachable mailbox for one attempt
504    /// proves nothing about the others: any busy mailbox is reported as busy,
505    /// any attempt still waiting yields `false`, and only when every mailbox
506    /// failed is the latest attempt's error reported. Once a response is
507    /// pinned, only that attempt is checked.
508    pub async fn complete(&self, store: &ProtectedLocalKeyStore, db: &Database) -> Result<bool> {
509        let attempts = store.peer_attempts(db, &self.locator).await?;
510        let (mut busy, mut waiting, mut failed) = (None, false, None);
511        for peer in attempts.iter().rev() {
512            match self
513                .exchange(
514                    Operation::Mailbox {
515                        vault: peer.vault(),
516                        handle: peer.handle(),
517                    },
518                    None,
519                )
520                .await
521            {
522                Ok(Reply::Mailbox(mail)) if mail.admission.is_some() => {
523                    let grant = store.open_peer_response(db, &mail).await?;
524                    self.finish(store, db, peer, &mail, grant).await?;
525                    return Ok(true);
526                }
527                Ok(Reply::Mailbox(_)) => waiting = true,
528                Ok(_) => {
529                    failed.get_or_insert(anyhow::anyhow!("error enrollment-response"));
530                }
531                Err(error) if super::errors::has_code(&error, "enrollment-busy") => {
532                    busy = Some(error)
533                }
534                Err(error) => {
535                    failed.get_or_insert(error);
536                }
537            }
538        }
539        match (busy, failed) {
540            (Some(error), _) => Err(error),
541            _ if waiting => Ok(false),
542            (None, Some(error)) => Err(error),
543            (None, None) => Ok(false),
544        }
545    }
546    pub async fn finish(
547        &self,
548        store: &ProtectedLocalKeyStore,
549        db: &Database,
550        peer: &membership::Joiner,
551        mail: &membership::Mailbox,
552        grant: membership::ProvisionalGrant,
553    ) -> Result<()> {
554        let context = Context {
555            vault: peer.vault(),
556            genesis: grant.genesis,
557            device: peer.device(),
558            head: grant.outcome,
559        };
560        // `finish_peer` verifies the chain while locating the outcome.
561        let evidence = self.unverified_membership(&context, peer.bearer()).await?;
562        store.finish_peer(db, mail, &evidence).await
563    }
564}
aven_core/sync/client/enrollment.rs

aven_core/sync/client/
enrollment.rs
