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

aven_core/sync/client/
enrollment.rs
