coven_replication/sync/store/device_join/authorized_join/
admission.rs

1use super::*;
2
3impl<'operation, 'storage> AuthorizedJoin<'operation, 'storage> {
4    fn sign_device_admission_approval(
5        &self,
6        request: DeviceProviderAccessRequest,
7        admission: DeviceProviderAdmission,
8    ) -> Result<DeviceProviderAdmissionApproval, DeviceJoinError> {
9        self.local_writer
10            .sign_device_admission_approval(request, admission, &self.verified_root)
11    }
12
13    async fn activate(
14        &mut self,
15        batch: crate::sync::store::commit_publication::operation::commit_plan::StoreOperationBatch,
16    ) -> Result<StoreBatchCommitRef, crate::sync::store::StoreError> {
17        let plan = self.writer.prepare_plan().await?;
18        self.writer.activate(plan, batch).await
19    }
20
21    async fn publish_cross_principal_challenge(
22        &mut self,
23        authorization: &DeviceJoinChallengePublicationAuthorization,
24        challenge: &CrossPrincipalProbeChallenge,
25        context: &coven_protocol::provider::CrossPrincipalChallengeContext,
26        store: &StoreProviderBinding,
27        attempt_owner: &StoreDeviceRegistration,
28    ) -> Result<CrossPrincipalProbeChallenge, DeviceJoinError> {
29        self.local_writer
30            .verify_cross_principal_challenge(challenge, context, store)
31            .map_err(DeviceJoinError::ProviderProbe)?;
32        if authorization.attempt_id != context.attempt_id {
33            return Err(DeviceJoinError::AttemptMismatch);
34        }
35        // The commit that opened the attempt is what the challenge is
36        // authorized against; there is no separate attempt file to agree with.
37        let activation = self
38            .join_history()
39            .load_commit(&authorization.attempt_activation)
40            .await?;
41        if activation.author() != attempt_owner
42            || !activation
43                .device_join_attempt_decisions()
44                .iter()
45                .any(|decision| {
46                    matches!(
47                        decision,
48                        DeviceJoinAttemptDecisionRef::Attempt(opened)
49                            if *opened == authorization.attempt_id
50                    )
51                })
52        {
53            return Err(DeviceJoinError::AttemptMismatch);
54        }
55        self.storage
56            .settle_cross_principal_challenge(
57                &self.database,
58                authorization,
59                challenge,
60                context,
61                store,
62            )
63            .await
64            .map_err(DeviceJoinError::ProviderProbe)
65    }
66
67    pub(crate) async fn authorize_access(
68        &mut self,
69        request: DeviceProviderAccessRequest,
70        access_administrator: Option<&dyn DeviceProviderAccessAdministrator>,
71    ) -> Result<DeviceProviderAdmissionApproval, DeviceJoinError> {
72        let provider_admin = self.resolve_provider_admin(&request.offer.provider_admin.grant_id)?;
73        if provider_admin != *request.offer.provider_admin {
74            return Err(DeviceJoinError::OfferMismatch);
75        }
76        let owner = self
77            .join_history()
78            .load_registration(&request.offer.owner_registration)
79            .await?
80            .value;
81        request.verify(&owner)?;
82        if !self
83            .local_writer
84            .is_authored_by_registration(&provider_admin.administrator)
85        {
86            return Err(DeviceJoinError::ProviderAdministratorRequired);
87        }
88        let database = self.database.clone();
89        let journal = self.journal(request.offer.attempt_id);
90        let current = journal.current().await?;
91        let durable = match &*current.progress {
92            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Offered(_)) => {
93                journal
94                    .advance(
95                        &current,
96                        OwnerJoinProgress::AccessRequested(request.clone()),
97                    )
98                    .await?
99            }
100            _ => current,
101        };
102        let initial = durable.clone();
103        if provider_admin.provider == request.peer_provider {
104            return match &*durable.progress {
105                DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ApprovalPrepared(approval)) => {
106                    Ok(approval.clone())
107                }
108                DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AccessRequested(
109                    durable_request,
110                )) if durable_request == &request => {
111                    let approval = self.sign_device_admission_approval(
112                        request,
113                        DeviceProviderAdmission::SamePrincipal,
114                    )?;
115                    journal
116                        .advance(
117                            &durable,
118                            OwnerJoinProgress::ApprovalPrepared(approval.clone()),
119                        )
120                        .await?;
121                    Ok(approval)
122                }
123                _ => Err(DeviceJoinError::JournalConflict),
124            };
125        }
126        let (grant, prepared, prepared_progress) = match &*durable.progress {
127            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ApprovalPrepared(approval)) => {
128                return Ok(approval.clone())
129            }
130            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AccessGrantPrepared {
131                request: durable_request,
132                grant,
133                prepared,
134            }) if durable_request == &request => {
135                (grant.clone(), prepared.restore()?, durable.clone())
136            }
137            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AccessRequested(durable_request))
138                if durable_request == &request =>
139            {
140                let administrator =
141                    access_administrator.ok_or(DeviceJoinError::ProviderAdministratorRequired)?;
142                let locator = administrator
143                    .grant_member_access(
144                        &request.offer.member_pubkey,
145                        self.membership
146                            .current_member_provider_email(&request.offer.member_pubkey),
147                        &request.peer_provider,
148                    )
149                    .await?;
150                let grant_id = ProviderAccessGrantId::from_random_bytes(
151                    *ObjectHash::digest(database.new_store_write_id().as_str().as_bytes())
152                        .as_bytes(),
153                );
154                let grant = self
155                    .local_writer
156                    .sign_provider_access_grant(
157                        grant_id,
158                        request.offer.member_pubkey.clone(),
159                        request.peer_provider.clone(),
160                        locator,
161                        provider_admin.grant_id.clone(),
162                        provider_admin.administrator.clone(),
163                        &request.offer.provider,
164                    )
165                    .map_err(DeviceJoinError::ProviderProbe)?;
166                let context = coven_protocol::objects::ProtocolObjectContext::signed_plaintext(
167                    request.offer.store_root.store_root_hash,
168                    ProtocolObjectDomain::ProviderAccessGrant,
169                );
170                let prefix = coven_protocol::store_commit::provider_access_grant_semantic_prefix(
171                    &grant.grant_id,
172                );
173                let slot = self
174                    .storage
175                    .allocate_protocol_slot(&context, &prefix, ".json")
176                    .await?;
177                let prepared = self.storage.prepare_protocol_object(
178                    &context,
179                    slot,
180                    &prefix,
181                    grant.to_bytes(),
182                )?;
183                let prepared_progress = journal
184                    .advance(
185                        &initial,
186                        OwnerJoinProgress::AccessGrantPrepared {
187                            request: request.clone(),
188                            grant: grant.clone(),
189                            prepared: PreparedDeviceJoinObject::from_prepared(&prepared),
190                        },
191                    )
192                    .await?;
193                (grant, prepared, prepared_progress)
194            }
195            _ => return Err(DeviceJoinError::JournalConflict),
196        };
197        let context = coven_protocol::objects::ProtocolObjectContext::signed_plaintext(
198            request.offer.store_root.store_root_hash,
199            ProtocolObjectDomain::ProviderAccessGrant,
200        );
201        let prefix =
202            coven_protocol::store_commit::provider_access_grant_semantic_prefix(&grant.grant_id);
203        self.storage
204            .create_verified_protocol_object(&context, &prepared, &prefix, &grant.to_bytes())
205            .await
206            .map_err(|error| {
207                DeviceJoinError::prepared_object(
208                    error,
209                    DeviceJoinError::Provider(
210                        "provider access grant prepared object differs from its signed bytes"
211                            .to_string(),
212                    ),
213                )
214            })?;
215        let grant_ref =
216            StoreMemberProviderAccessGrantRef::from_grant(&grant, prepared.reference().clone());
217        let activation = self
218            .activate(
219                crate::sync::store::commit_publication::operation::commit_plan::StoreOperationBatch::ProviderAccessGrant(
220                    grant_ref.clone(),
221                ),
222            )
223            .await?;
224        let challenge_context = request.cross_challenge_context();
225        let probe_id = coven_protocol::provider::ProviderProbeId::from_bytes(
226            *ObjectHash::digest(database.new_store_write_id().as_str().as_bytes()).as_bytes(),
227        );
228        let challenge = self
229            .storage
230            .prepare_cross_principal_challenge(
231                &database,
232                probe_id,
233                &request.offer.provider,
234                &challenge_context,
235                self.local_writer.as_ref(),
236            )
237            .await
238            .map_err(DeviceJoinError::ProviderProbe)?;
239        let approval = self.sign_device_admission_approval(
240            request,
241            DeviceProviderAdmission::CrossPrincipal {
242                access_grant: Box::new(ActivatedStoreMemberProviderAccessGrant {
243                    grant,
244                    grant_ref,
245                    activation,
246                }),
247                challenge,
248            },
249        )?;
250        journal
251            .advance(
252                &prepared_progress,
253                OwnerJoinProgress::ApprovalPrepared(approval.clone()),
254            )
255            .await?;
256        Ok(approval)
257    }
258
259    pub(crate) async fn publish_challenge(
260        &mut self,
261        bootstrap: ProvisionalDeviceBootstrap,
262    ) -> Result<ProviderReadyDeviceBootstrap, DeviceJoinError> {
263        let offer = &bootstrap.request.approval().request.offer;
264        if &self.resolve_provider_admin(&offer.provider_admin.grant_id)?
265            != offer.provider_admin.as_ref()
266        {
267            return Err(DeviceJoinError::OfferMismatch);
268        }
269        let owner = self
270            .join_history()
271            .load_registration(&offer.owner_registration)
272            .await?
273            .value;
274        self.local_writer.verify_own_device_admission_approval(
275            bootstrap.request.approval(),
276            &self.verified_root,
277        )?;
278        let challenge_publication = match &bootstrap.request.approval().admission {
279            DeviceProviderAdmission::SamePrincipal => {
280                DeviceProviderChallengePublication::SamePrincipal
281            }
282            DeviceProviderAdmission::CrossPrincipal { challenge, .. } => {
283                let context = bootstrap
284                    .request
285                    .approval()
286                    .request
287                    .cross_challenge_context();
288                let authorization = DeviceJoinChallengePublicationAuthorization {
289                    attempt_id: bootstrap.publication_authorization.attempt_id,
290                    attempt_activation: bootstrap
291                        .publication_authorization
292                        .attempt_activation
293                        .clone(),
294                };
295                let published = self
296                    .publish_cross_principal_challenge(
297                        &authorization,
298                        challenge,
299                        &context,
300                        &offer.provider,
301                        &owner,
302                    )
303                    .await?;
304                DeviceProviderChallengePublication::CrossPrincipal {
305                    challenge: published,
306                }
307            }
308        };
309        let attempt_id = offer.attempt_id;
310        let ready = ProviderReadyDeviceBootstrap {
311            bootstrap: Box::new(bootstrap),
312            challenge_publication,
313        };
314        let journal = self.journal(attempt_id);
315        let current = journal.current().await?;
316        match &*current.progress {
317            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ProviderReady(existing))
318                if existing == &ready =>
319            {
320                return Ok(ready)
321            }
322            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AttemptActivated(bootstrap))
323                if bootstrap == ready.bootstrap.as_ref() =>
324            {
325                let intent = journal
326                    .advance(
327                        &current,
328                        OwnerJoinProgress::ChallengeCreateIntent(*ready.bootstrap.clone()),
329                    )
330                    .await?;
331                journal
332                    .advance(&intent, OwnerJoinProgress::ProviderReady(ready.clone()))
333                    .await?;
334            }
335            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ChallengeCreateIntent(bootstrap))
336                if bootstrap == ready.bootstrap.as_ref() =>
337            {
338                journal
339                    .advance(&current, OwnerJoinProgress::ProviderReady(ready.clone()))
340                    .await?;
341            }
342            _ => return Err(DeviceJoinError::JournalConflict),
343        }
344        Ok(ready)
345    }
346
347    pub(super) async fn complete_admission(
348        &mut self,
349        readiness: DeviceJoinReadiness,
350    ) -> Result<DeviceProviderAdmissionCompletion, DeviceJoinError> {
351        let attempt_id = readiness.proof.attempt_id;
352        let database = self.database.clone();
353        let journal = self.journal(attempt_id);
354        let current = journal.current().await?;
355        if let DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Completed(existing)) =
356            &*current.progress
357        {
358            if matches!(
359                existing,
360                DeviceProviderAdmissionCompletion::CrossPrincipal {
361                    readiness: durable,
362                    ..
363                } if **durable == readiness
364            ) {
365                return Ok(existing.clone());
366            }
367            return Err(DeviceJoinError::JournalConflict);
368        }
369        let bootstrap = match &*current.progress {
370            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ProviderReady(bootstrap)) => {
371                bootstrap.clone()
372            }
373            _ => return Err(DeviceJoinError::JournalConflict),
374        };
375        if readiness.proof.attempt_id != bootstrap.bootstrap.publication_authorization.attempt_id {
376            return Err(DeviceJoinError::AttemptMismatch);
377        }
378        let offer = &bootstrap.bootstrap.request.approval().request.offer;
379        let provider_admin = self.resolve_provider_admin(&offer.provider_admin.grant_id)?;
380        if &provider_admin != offer.provider_admin.as_ref() {
381            return Err(DeviceJoinError::ProviderAdministratorRequired);
382        }
383        let receipt = match (
384            &bootstrap.bootstrap.request.approval().admission,
385            &bootstrap.bootstrap.request.response(),
386            &readiness.provider,
387        ) {
388            (
389                DeviceProviderAdmission::CrossPrincipal { challenge, .. },
390                DeviceProviderResponseReservation::CrossPrincipal { response_slot },
391                DeviceProviderReadiness::CrossPrincipal(response),
392            ) => {
393                let context = coven_protocol::provider::CrossPrincipalResponseContext {
394                    challenge: bootstrap
395                        .bootstrap
396                        .request
397                        .approval()
398                        .request
399                        .cross_challenge_context(),
400                    expected_registration_hash: bootstrap
401                        .bootstrap
402                        .request
403                        .expected_registration()
404                        .registration_hash(),
405                    response_slot: response_slot.clone(),
406                };
407                self.storage
408                    .complete_cross_principal_probe(
409                        &database,
410                        challenge,
411                        response,
412                        &context,
413                        &offer.provider,
414                        self.local_writer.as_ref(),
415                        &offer.member_pubkey,
416                    )
417                    .await
418                    .map_err(DeviceJoinError::ProviderProbe)?
419            }
420            _ => return Err(DeviceJoinError::AttemptMismatch),
421        };
422        let completion = DeviceProviderAdmissionCompletion::CrossPrincipal {
423            bootstrap: Box::new(bootstrap.clone()),
424            readiness: Box::new(readiness.clone()),
425            receipt,
426        };
427        let observed = journal
428            .advance(&current, OwnerJoinProgress::ResponseObserved(readiness))
429            .await?;
430        journal
431            .advance(&observed, OwnerJoinProgress::Completed(completion.clone()))
432            .await?;
433        Ok(completion)
434    }
435
436    pub(crate) async fn complete_same_principal(
437        &mut self,
438        bootstrap: ProviderReadyDeviceBootstrap,
439    ) -> Result<DeviceProviderAdmissionCompletion, DeviceJoinError> {
440        let attempt_id = bootstrap.bootstrap.publication_authorization.attempt_id;
441        if !matches!(
442            (
443                &bootstrap.bootstrap.request.approval().admission,
444                bootstrap.bootstrap.request.response(),
445                &bootstrap.challenge_publication,
446            ),
447            (
448                DeviceProviderAdmission::SamePrincipal,
449                DeviceProviderResponseReservation::SamePrincipal,
450                DeviceProviderChallengePublication::SamePrincipal,
451            )
452        ) {
453            return Err(DeviceJoinError::AttemptMismatch);
454        }
455        let journal = self.journal(attempt_id);
456        let current = journal.current().await?;
457        if let DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Completed(existing)) =
458            &*current.progress
459        {
460            return match existing {
461                DeviceProviderAdmissionCompletion::SamePrincipal { bootstrap: durable }
462                    if **durable == bootstrap =>
463                {
464                    Ok(existing.clone())
465                }
466                _ => Err(DeviceJoinError::JournalConflict),
467            };
468        }
469        match &*current.progress {
470            DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ProviderReady(durable))
471                if durable == &bootstrap => {}
472            _ => return Err(DeviceJoinError::JournalConflict),
473        }
474        let completion = DeviceProviderAdmissionCompletion::SamePrincipal {
475            bootstrap: Box::new(bootstrap),
476        };
477        journal
478            .advance(&current, OwnerJoinProgress::Completed(completion.clone()))
479            .await?;
480        Ok(completion)
481    }
482}
483
484impl Store {
485    #[doc(hidden)]
486    pub(crate) async fn authorize_device_provider_access(
487        &self,
488        request: DeviceProviderAccessRequest,
489        access_administrator: Option<&dyn DeviceProviderAccessAdministrator>,
490    ) -> Result<DeviceProviderAdmissionApproval, DeviceJoinError> {
491        let mut writer = self
492            .authorize_writer()
493            .await
494            .map_err(DeviceJoinError::from)?;
495        writer
496            .join_operation()
497            .authorize_access(request, access_administrator)
498            .await
499    }
500
501    #[doc(hidden)]
502    pub(crate) async fn publish_device_provider_challenge(
503        &self,
504        bootstrap: ProvisionalDeviceBootstrap,
505    ) -> Result<ProviderReadyDeviceBootstrap, DeviceJoinError> {
506        let mut writer = self
507            .authorize_writer()
508            .await
509            .map_err(DeviceJoinError::from)?;
510        writer.join_operation().publish_challenge(bootstrap).await
511    }
512
513    #[doc(hidden)]
514    pub(crate) async fn complete_device_provider_admission(
515        &self,
516        readiness: DeviceJoinReadiness,
517    ) -> Result<DeviceProviderAdmissionCompletion, DeviceJoinError> {
518        let mut writer = self
519            .authorize_writer()
520            .await
521            .map_err(DeviceJoinError::from)?;
522        writer.join_operation().complete_admission(readiness).await
523    }
524
525    #[doc(hidden)]
526    pub(crate) async fn complete_same_principal_device_admission(
527        &self,
528        bootstrap: ProviderReadyDeviceBootstrap,
529    ) -> Result<DeviceProviderAdmissionCompletion, DeviceJoinError> {
530        let mut writer = self
531            .authorize_writer()
532            .await
533            .map_err(DeviceJoinError::from)?;
534        writer
535            .join_operation()
536            .complete_same_principal(bootstrap)
537            .await
538    }
539}
540
541#[async_trait::async_trait]
542pub trait DeviceProviderAccessAdministrator: Send + Sync {
543    async fn grant_member_access(
544        &self,
545        member_pubkey: &str,
546        provider_account_email: Option<&str>,
547        peer: &ProviderDeviceBinding,
548    ) -> Result<coven_protocol::provider::ProviderAccessLocator, DeviceJoinError>;
549}
coven_replication/sync/store/device_join/authorized_join/admission.rs

coven_replication/sync/store/device_join/authorized_join/
admission.rs
